abhishekbafna commented on code in PR #15515:
URL: https://github.com/apache/pinot/pull/15515#discussion_r2059556680
##########
pinot-common/src/main/java/org/apache/pinot/common/config/provider/TableCache.java:
##########
@@ -390,6 +423,78 @@ private List<Schema> getSchemas() {
return schemas;
}
+ private void addLogicalTables(List<String> paths) {
+ // Subscribe data changes before reading the data to avoid missing changes
+ for (String path : paths) {
+ _propertyStore.subscribeDataChanges(path, _zkLogicalTableChangeListener);
+ }
+ List<ZNRecord> znRecords = _propertyStore.get(paths, null,
AccessOption.PERSISTENT);
+ for (ZNRecord znRecord : znRecords) {
+ if (znRecord != null) {
+ try {
+ putLogicalTable(znRecord);
+ } catch (Exception e) {
+ LOGGER.error("Caught exception while adding logical table for
ZNRecord: {}", znRecord.getId(), e);
+ }
+ }
+ }
+ }
+
+ private void putLogicalTable(ZNRecord znRecord)
+ throws IOException {
+ LogicalTable logicalTable = LogicalTableUtils.fromZNRecord(znRecord);
+ String logicalTableName = logicalTable.getTableName();
+ _logicalTableNameMap.put(logicalTableName, logicalTable);
+ }
+
+ private void removeLogicalTable(String path) {
+ _propertyStore.unsubscribeDataChanges(path, _zkLogicalTableChangeListener);
+ String logicalTableName =
path.substring(LOGICAL_TABLE_PATH_PREFIX.length());
+ _logicalTableNameMap.remove(logicalTableName);
+ }
+
+ private void notifyLogicalTableChangeListeners() {
+ if (!_logicalTableChangeListeners.isEmpty()) {
+ List<LogicalTable> logicalTables = new
ArrayList<>(_logicalTableNameMap.values());
+ for (LogicalTableChangeListener listener : _logicalTableChangeListeners)
{
+ listener.onChange(logicalTables);
+ }
+ }
+ }
+
+ @Nullable
+ public LogicalTable getLogicalTable(String logicalTableName) {
+ return _logicalTableNameMap.get(logicalTableName);
+ }
+
+ public boolean isLogicalTable(String logicalTableName) {
+ return _logicalTableNameMap.containsKey(logicalTableName);
+ }
+
+ public List<LogicalTable> getLogicalTables() {
+ return new ArrayList<>(_logicalTableNameMap.values());
+ }
+
+ public Schema getLogicalTableSchema(String logicalTableName) {
Review Comment:
> We want to first lookup the schema using raw logicalTableName, then
fallback to the first physical table.
This would helpful when logical table has a schema enforced by the system.
Otherwise we would not find schema same name as logical table.
We can consider having an optional field to accept the schema from the user
otherwise fallback to the first physical table schema.
@vrajat please check and comment.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
