ahmedabu98 commented on code in PR #39883:
URL: https://github.com/apache/beam/pull/39883#discussion_r3937620859
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/TableMetadataDriver.java:
##########
@@ -337,7 +382,57 @@ public void processElement(
"Table '{}' does not exist in catalog. Skipping metadata emission
for side-input view.",
tableIdString);
TABLES_SKIPPED_MISSING_COUNTER.inc();
+ } catch (IllegalArgumentException e) {
+ LOG.warn(
+ "Failed to parse table identifier '{}'. Skipping metadata emission
for side-input view.",
+ tableIdString,
+ e);
+ TABLES_SKIPPED_MISSING_COUNTER.inc();
}
}
}
+
+ static class AccumulateTableMetadataMapDoFn
+ extends DoFn<
+ KV<Void, KV<String, SerializableTableSpec>>, Map<String,
SerializableTableSpec>> {
+ @StateId("tableCache")
+ private final StateSpec<MapState<String, SerializableTableSpec>>
cacheStateSpec =
+ StateSpecs.map(StringUtf8Coder.of(), SerializableTableSpec.getCoder());
+
+ @ProcessElement
+ public void processElement(
+ @Element KV<Void, KV<String, SerializableTableSpec>> element,
+ @StateId("tableCache") MapState<String, SerializableTableSpec>
cacheState,
+ OutputReceiver<Map<String, SerializableTableSpec>> out) {
+ KV<String, SerializableTableSpec> kv = element.getValue();
+ cacheState.put(kv.getKey(), kv.getValue());
+
+ Map<String, SerializableTableSpec> mapSnapshot = new HashMap<>();
+ for (Map.Entry<String, SerializableTableSpec> entry :
cacheState.entries().read()) {
+ mapSnapshot.put(entry.getKey(), entry.getValue());
+ }
+ out.output(Collections.unmodifiableMap(mapSnapshot));
+ }
+ }
+
+ static class MapMergerFn extends Combine.BinaryCombineFn<Map<String,
SerializableTableSpec>> {
+ @Override
+ public Map<String, SerializableTableSpec> apply(
+ Map<String, SerializableTableSpec> left, Map<String,
SerializableTableSpec> right) {
+ if (left == null || left.isEmpty()) {
+ return right != null ? right : Collections.emptyMap();
+ }
+ if (right == null || right.isEmpty()) {
+ return left;
+ }
+ Map<String, SerializableTableSpec> merged = new HashMap<>(left);
+ merged.putAll(right);
Review Comment:
Are we sure that `right` will always be the newer metadata?
I think we should store `TableMetadata.lastUpdatedMillis()` in
`SerializableTableSpec` and use it to compare `right` and `left` to choose the
freshest metadata
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/TableMetadataDriver.java:
##########
@@ -337,7 +382,57 @@ public void processElement(
"Table '{}' does not exist in catalog. Skipping metadata emission
for side-input view.",
tableIdString);
TABLES_SKIPPED_MISSING_COUNTER.inc();
+ } catch (IllegalArgumentException e) {
+ LOG.warn(
+ "Failed to parse table identifier '{}'. Skipping metadata emission
for side-input view.",
+ tableIdString,
+ e);
+ TABLES_SKIPPED_MISSING_COUNTER.inc();
}
}
}
+
+ static class AccumulateTableMetadataMapDoFn
+ extends DoFn<
+ KV<Void, KV<String, SerializableTableSpec>>, Map<String,
SerializableTableSpec>> {
+ @StateId("tableCache")
+ private final StateSpec<MapState<String, SerializableTableSpec>>
cacheStateSpec =
+ StateSpecs.map(StringUtf8Coder.of(), SerializableTableSpec.getCoder());
+
+ @ProcessElement
+ public void processElement(
+ @Element KV<Void, KV<String, SerializableTableSpec>> element,
+ @StateId("tableCache") MapState<String, SerializableTableSpec>
cacheState,
+ OutputReceiver<Map<String, SerializableTableSpec>> out) {
+ KV<String, SerializableTableSpec> kv = element.getValue();
+ cacheState.put(kv.getKey(), kv.getValue());
+
+ Map<String, SerializableTableSpec> mapSnapshot = new HashMap<>();
+ for (Map.Entry<String, SerializableTableSpec> entry :
cacheState.entries().read()) {
+ mapSnapshot.put(entry.getKey(), entry.getValue());
Review Comment:
I think state cache is a good idea, but we should try clean up old tables
otherwise state and side-input will get unnecessarily bloated.
Might help to store a separate "last seen" millis next to each table, and
renew it every time a new table comes thru here. We can use it to identify idle
tables (older than `refreshInterval`) and remove them from the cache
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/TableMetadataDriver.java:
##########
@@ -337,7 +382,57 @@ public void processElement(
"Table '{}' does not exist in catalog. Skipping metadata emission
for side-input view.",
tableIdString);
TABLES_SKIPPED_MISSING_COUNTER.inc();
+ } catch (IllegalArgumentException e) {
+ LOG.warn(
+ "Failed to parse table identifier '{}'. Skipping metadata emission
for side-input view.",
+ tableIdString,
+ e);
+ TABLES_SKIPPED_MISSING_COUNTER.inc();
Review Comment:
Can we break up this try block into separate blocks?
`SerializableTableSpec.fromTable` can throw `IllegalArgumentException`, which
this code would log as a table identifier parsing error
--
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]
