You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "aiborodin (via GitHub)" <gi...@apache.org> on 2025/10/23 08:11:13 UTC
[PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
aiborodin opened a new pull request, #14406:
URL: https://github.com/apache/iceberg/pull/14406
The `TableMetadataCache` in `DynamicIcebergSink` should only refresh after it exists longer than a configured refresh interval. Not the opposite =)
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2454486845
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -50,13 +51,19 @@ class TableMetadataCache {
private final Catalog catalog;
private final long refreshMs;
+ private final Clock cacheRefreshClock;
private final int inputSchemasPerTableCacheMaximumSize;
private final Map<TableIdentifier, CacheItem> tableCache;
TableMetadataCache(
- Catalog catalog, int maximumSize, long refreshMs, int inputSchemasPerTableCacheMaximumSize) {
+ Catalog catalog,
+ int maximumSize,
+ long refreshMs,
+ Clock cacheRefreshClock,
Review Comment:
Could we just create a separate constructor for testing, and annotate with `@VisibleForTesting`?
I usually try to separate out the testing only code from the prod code as much as possible, and needing a `Clock` in the constructor is a bit confusing
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2587876524
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -186,14 +202,15 @@ private Tuple2<Boolean, Exception> refreshTable(TableIdentifier identifier) {
return EXISTS;
} catch (NoSuchTableException e) {
LOG.debug("Table doesn't exist {}", identifier, e);
- tableCache.put(identifier, new CacheItem(false, null, null, null, 1));
+ tableCache.put(identifier, new CacheItem(-1, false, null, null, null, 1));
return Tuple2.of(false, e);
}
}
private boolean needsRefresh(CacheItem cacheItem, boolean allowRefresh) {
return allowRefresh
- && (cacheItem == null || cacheItem.created + refreshMs > System.currentTimeMillis());
+ && (cacheItem == null
+ || refreshMs < cacheRefreshClock.millis() - cacheItem.createdTimestampMillis);
Review Comment:
The new condition `cacheRefreshClock.millis() - cacheItem.createdTimestampMillis > refreshMs` seems more natural and easier to read. It reads to me as "elapsed time greater than the refresh interval".
While the existing one (with the sign changed) seems less intuitive.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#issuecomment-3616159699
@aiborodin: Do we want to see this in 1.10.1?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#issuecomment-3623739446
Thank you for the review @pvary!
Yes, let's include this in 1.10.1.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2454474949
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -80,6 +87,14 @@ ResolvedSchemaInfo schema(TableIdentifier identifier, Schema input) {
return schema(identifier, input, true);
}
+ ResolvedSchemaInfo getResolvedSchemaInfo(TableIdentifier identifier, Schema input) {
+ CacheItem tableCacheItem = tableCache.get(identifier);
+ if (tableCacheItem == null) {
+ return null;
+ }
+ return tableCacheItem.inputSchemas.get(input);
+ }
+
Review Comment:
Why is this method?
Is it for testing only?
Could we just reuse an existing one?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2468780773
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -50,13 +51,19 @@ class TableMetadataCache {
private final Catalog catalog;
private final long refreshMs;
+ private final Clock cacheRefreshClock;
Review Comment:
If we update the cache in `TableMetadataCache`, do we also need to sync the cache inside `DynamicWriteResultAggregator` and `DynamicWriter`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#issuecomment-3615732513
Merged to main.
Thanks for the fix @aiborodin!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2591269682
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -50,13 +51,19 @@ class TableMetadataCache {
private final Catalog catalog;
private final long refreshMs;
+ private final Clock cacheRefreshClock;
private final int inputSchemasPerTableCacheMaximumSize;
private final Map<TableIdentifier, CacheItem> tableCache;
TableMetadataCache(
- Catalog catalog, int maximumSize, long refreshMs, int inputSchemasPerTableCacheMaximumSize) {
+ Catalog catalog,
+ int maximumSize,
+ long refreshMs,
+ Clock cacheRefreshClock,
Review Comment:
Acknowledged.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2587879449
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -50,13 +51,19 @@ class TableMetadataCache {
private final Catalog catalog;
private final long refreshMs;
+ private final Clock cacheRefreshClock;
Review Comment:
I think this is unnecessary. Caches in `DynamicWriteResultAggregator` and `DynamicWriter` should adjust to incoming updates.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2454479675
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -186,14 +202,15 @@ private Tuple2<Boolean, Exception> refreshTable(TableIdentifier identifier) {
return EXISTS;
} catch (NoSuchTableException e) {
LOG.debug("Table doesn't exist {}", identifier, e);
- tableCache.put(identifier, new CacheItem(false, null, null, null, 1));
+ tableCache.put(identifier, new CacheItem(-1, false, null, null, null, 1));
Review Comment:
This is not good.
We intentionally don't want to flood the catalog with requests even on missing tables.
We want to hold back requests there too, and only check after the cache timeout agan. That is why we store the CacheItem with the current time
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2591872855
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -186,14 +200,16 @@ private Tuple2<Boolean, Exception> refreshTable(TableIdentifier identifier) {
return EXISTS;
} catch (NoSuchTableException e) {
LOG.debug("Table doesn't exist {}", identifier, e);
- tableCache.put(identifier, new CacheItem(false, null, null, null, 1));
+ tableCache.put(
+ identifier, new CacheItem(cacheRefreshClock.millis(), false, null, null, null, 1));
return Tuple2.of(false, e);
}
}
private boolean needsRefresh(CacheItem cacheItem, boolean allowRefresh) {
return allowRefresh
- && (cacheItem == null || cacheItem.created + refreshMs > System.currentTimeMillis());
+ && (cacheItem == null
+ || cacheRefreshClock.millis() - cacheItem.createdTimestampMillis > refreshMs);
Review Comment:
@aiborodin Good catch. Thanks for this fix!
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2454741721
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java:
##########
@@ -87,8 +94,71 @@ void testCachingDisabled() {
Catalog catalog = CATALOG_EXTENSION.catalog();
TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
catalog.createTable(tableIdentifier, SCHEMA);
- TableMetadataCache cache = new TableMetadataCache(catalog, 0, Long.MAX_VALUE, 10);
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 0, Long.MAX_VALUE, Clock.systemUTC(), 10);
assertThat(cache.getInternalCache()).isEmpty();
}
+
+ @Test
+ void testNoCacheRefreshBeforeRefreshIntervalElapses() {
+ // Create table
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA2);
+
+ // Create test clock
+ long firstEpochMillis = 1L;
+ MutableClock testClock = new MutableClock(ZoneOffset.UTC, firstEpochMillis);
+
+ // Init cache
+ long refreshIntervalMillis = 100L;
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, refreshIntervalMillis, testClock, 10);
+ cache.update(tableIdentifier, table);
+
+ // Cache schema
+ Schema schema = cache.schema(tableIdentifier, SCHEMA2).resolvedTableSchema();
+ assertThat(schema.sameSchema(SCHEMA2)).isTrue();
+
+ // Progress clock to timestamp before refresh interval
+ long secondEpochMilli = 2L;
+ testClock.setEpochMilli(secondEpochMilli);
+
+ // Cache schema with fewer fields
+ TableMetadataCache.ResolvedSchemaInfo schemaInfo = cache.schema(tableIdentifier, SCHEMA);
+ assertThat(schemaInfo.resolvedTableSchema().sameSchema(SCHEMA2)).isTrue();
+ assertThat(schemaInfo.compareResult())
+ .isEqualTo(CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED);
+
+ // Assert both schemas are in cache
+ assertThat(cache.getResolvedSchemaInfo(tableIdentifier, SCHEMA)).isNotNull();
+ assertThat(cache.getResolvedSchemaInfo(tableIdentifier, SCHEMA2)).isNotNull();
+ }
+
+ private static class MutableClock extends Clock {
+ private final ZoneId zoneId;
+ private long epochMilli;
+
+ MutableClock(ZoneId zoneId, long epochMilli) {
+ this.zoneId = zoneId;
+ this.epochMilli = epochMilli;
+ }
+
+ void setEpochMilli(long epochMilli) {
+ this.epochMilli = epochMilli;
+ }
+
+ public ZoneId getZone() {
+ return this.zoneId;
+ }
+
+ public Clock withZone(ZoneId zone) {
+ return zone.equals(this.zoneId) ? this : new MutableClock(zone, this.epochMilli);
+ }
+
+ public Instant instant() {
Review Comment:
Please add `@Override` when appropriate.
Also, maybe only implement what is required for us
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2587877538
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java:
##########
@@ -87,8 +94,71 @@ void testCachingDisabled() {
Catalog catalog = CATALOG_EXTENSION.catalog();
TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
catalog.createTable(tableIdentifier, SCHEMA);
- TableMetadataCache cache = new TableMetadataCache(catalog, 0, Long.MAX_VALUE, 10);
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 0, Long.MAX_VALUE, Clock.systemUTC(), 10);
assertThat(cache.getInternalCache()).isEmpty();
}
+
+ @Test
+ void testNoCacheRefreshBeforeRefreshIntervalElapses() {
+ // Create table
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA2);
+
+ // Create test clock
+ long firstEpochMillis = 1L;
+ MutableClock testClock = new MutableClock(ZoneOffset.UTC, firstEpochMillis);
Review Comment:
I replaced this class with `Clock.fixed()`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#issuecomment-3590734568
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#issuecomment-3616330138
+1 Let's port this to 1.10.1.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2587906483
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -186,14 +202,15 @@ private Tuple2<Boolean, Exception> refreshTable(TableIdentifier identifier) {
return EXISTS;
} catch (NoSuchTableException e) {
LOG.debug("Table doesn't exist {}", identifier, e);
- tableCache.put(identifier, new CacheItem(false, null, null, null, 1));
+ tableCache.put(identifier, new CacheItem(-1, false, null, null, null, 1));
Review Comment:
Acknowledged, I added the current time.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2472091129
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -50,13 +51,19 @@ class TableMetadataCache {
private final Catalog catalog;
private final long refreshMs;
+ private final Clock cacheRefreshClock;
Review Comment:
Good question, I'll double check this.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2454469589
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -186,14 +202,15 @@ private Tuple2<Boolean, Exception> refreshTable(TableIdentifier identifier) {
return EXISTS;
} catch (NoSuchTableException e) {
LOG.debug("Table doesn't exist {}", identifier, e);
- tableCache.put(identifier, new CacheItem(false, null, null, null, 1));
+ tableCache.put(identifier, new CacheItem(-1, false, null, null, null, 1));
return Tuple2.of(false, e);
}
}
private boolean needsRefresh(CacheItem cacheItem, boolean allowRefresh) {
return allowRefresh
- && (cacheItem == null || cacheItem.created + refreshMs > System.currentTimeMillis());
+ && (cacheItem == null
+ || refreshMs < cacheRefreshClock.millis() - cacheItem.createdTimestampMillis);
Review Comment:
Could we just change the `>` to `<`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "pvary (via GitHub)" <gi...@apache.org>.
pvary commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2454742889
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java:
##########
@@ -87,8 +94,71 @@ void testCachingDisabled() {
Catalog catalog = CATALOG_EXTENSION.catalog();
TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
catalog.createTable(tableIdentifier, SCHEMA);
- TableMetadataCache cache = new TableMetadataCache(catalog, 0, Long.MAX_VALUE, 10);
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 0, Long.MAX_VALUE, Clock.systemUTC(), 10);
assertThat(cache.getInternalCache()).isEmpty();
}
+
+ @Test
+ void testNoCacheRefreshBeforeRefreshIntervalElapses() {
+ // Create table
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA2);
+
+ // Create test clock
+ long firstEpochMillis = 1L;
+ MutableClock testClock = new MutableClock(ZoneOffset.UTC, firstEpochMillis);
Review Comment:
Could we create a simple constructor. We likely don't use any of this
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2587876861
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -80,6 +87,14 @@ ResolvedSchemaInfo schema(TableIdentifier identifier, Schema input) {
return schema(identifier, input, true);
}
+ ResolvedSchemaInfo getResolvedSchemaInfo(TableIdentifier identifier, Schema input) {
+ CacheItem tableCacheItem = tableCache.get(identifier);
+ if (tableCacheItem == null) {
+ return null;
+ }
+ return tableCacheItem.inputSchemas.get(input);
+ }
+
Review Comment:
I replaced it with existing `@VisibleForTesting` methods in this PR.
Although I don't like these methods and would prefer hiding internal state behind simple `get()` calls.
We can do this in a separate PR.
I originally added this method to allow retrieving a cached schema without caching it if not present (which `TableMetadataCache.schema()` always does). Currently, this is only possible using "get internal cache" methods.
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -50,13 +51,19 @@ class TableMetadataCache {
private final Catalog catalog;
private final long refreshMs;
+ private final Clock cacheRefreshClock;
private final int inputSchemasPerTableCacheMaximumSize;
private final Map<TableIdentifier, CacheItem> tableCache;
TableMetadataCache(
- Catalog catalog, int maximumSize, long refreshMs, int inputSchemasPerTableCacheMaximumSize) {
+ Catalog catalog,
+ int maximumSize,
+ long refreshMs,
+ Clock cacheRefreshClock,
Review Comment:
It is a common pattern to inject a `Clock` as a dependency in Java (used for both prod and testing). Quoting [Java documentation](https://docs.oracle.com/en/java/javase/18/docs/api/java.base/java/time/Clock.html):
> Best practice for applications is to pass a Clock into any method that requires the current instant and time-zone. A dependency injection framework is one way to achieve this:
>
> ```java
> public class MyBean {
> private Clock clock; // dependency inject
> ...
> public void process(LocalDate eventDate) {
> if (eventDate.isBefore(LocalDate.now(clock)) {
> ...
> }
> }
> }
> ```
> This approach allows an alternative clock, such as [fixed](https://docs.oracle.com/en/java/javase/18/docs/api/java.base/java/time/Clock.html#fixed(java.time.Instant,java.time.ZoneId)) or [offset](https://docs.oracle.com/en/java/javase/18/docs/api/java.base/java/time/Clock.html#offset(java.time.Clock,java.time.Duration)) to be used during testing.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2596691303
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -186,14 +200,16 @@ private Tuple2<Boolean, Exception> refreshTable(TableIdentifier identifier) {
return EXISTS;
} catch (NoSuchTableException e) {
LOG.debug("Table doesn't exist {}", identifier, e);
- tableCache.put(identifier, new CacheItem(false, null, null, null, 1));
+ tableCache.put(
+ identifier, new CacheItem(cacheRefreshClock.millis(), false, null, null, null, 1));
return Tuple2.of(false, e);
}
}
private boolean needsRefresh(CacheItem cacheItem, boolean allowRefresh) {
return allowRefresh
- && (cacheItem == null || cacheItem.created + refreshMs > System.currentTimeMillis());
+ && (cacheItem == null
+ || cacheRefreshClock.millis() - cacheItem.createdTimestampMillis > refreshMs);
Review Comment:
Thank you @mxm.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2587903118
##########
flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/sink/dynamic/TestTableMetadataCache.java:
##########
@@ -87,8 +94,71 @@ void testCachingDisabled() {
Catalog catalog = CATALOG_EXTENSION.catalog();
TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
catalog.createTable(tableIdentifier, SCHEMA);
- TableMetadataCache cache = new TableMetadataCache(catalog, 0, Long.MAX_VALUE, 10);
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 0, Long.MAX_VALUE, Clock.systemUTC(), 10);
assertThat(cache.getInternalCache()).isEmpty();
}
+
+ @Test
+ void testNoCacheRefreshBeforeRefreshIntervalElapses() {
+ // Create table
+ Catalog catalog = CATALOG_EXTENSION.catalog();
+ TableIdentifier tableIdentifier = TableIdentifier.parse("default.myTable");
+ Table table = catalog.createTable(tableIdentifier, SCHEMA2);
+
+ // Create test clock
+ long firstEpochMillis = 1L;
+ MutableClock testClock = new MutableClock(ZoneOffset.UTC, firstEpochMillis);
+
+ // Init cache
+ long refreshIntervalMillis = 100L;
+ TableMetadataCache cache =
+ new TableMetadataCache(catalog, 10, refreshIntervalMillis, testClock, 10);
+ cache.update(tableIdentifier, table);
+
+ // Cache schema
+ Schema schema = cache.schema(tableIdentifier, SCHEMA2).resolvedTableSchema();
+ assertThat(schema.sameSchema(SCHEMA2)).isTrue();
+
+ // Progress clock to timestamp before refresh interval
+ long secondEpochMilli = 2L;
+ testClock.setEpochMilli(secondEpochMilli);
+
+ // Cache schema with fewer fields
+ TableMetadataCache.ResolvedSchemaInfo schemaInfo = cache.schema(tableIdentifier, SCHEMA);
+ assertThat(schemaInfo.resolvedTableSchema().sameSchema(SCHEMA2)).isTrue();
+ assertThat(schemaInfo.compareResult())
+ .isEqualTo(CompareSchemasVisitor.Result.DATA_CONVERSION_NEEDED);
+
+ // Assert both schemas are in cache
+ assertThat(cache.getResolvedSchemaInfo(tableIdentifier, SCHEMA)).isNotNull();
+ assertThat(cache.getResolvedSchemaInfo(tableIdentifier, SCHEMA2)).isNotNull();
+ }
+
+ private static class MutableClock extends Clock {
+ private final ZoneId zoneId;
+ private long epochMilli;
+
+ MutableClock(ZoneId zoneId, long epochMilli) {
+ this.zoneId = zoneId;
+ this.epochMilli = epochMilli;
+ }
+
+ void setEpochMilli(long epochMilli) {
+ this.epochMilli = epochMilli;
+ }
+
+ public ZoneId getZone() {
+ return this.zoneId;
+ }
+
+ public Clock withZone(ZoneId zone) {
+ return zone.equals(this.zoneId) ? this : new MutableClock(zone, this.epochMilli);
+ }
+
+ public Instant instant() {
Review Comment:
I replaced this class with `Clock.fixed()`.
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
Guosmilesmile commented on code in PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#discussion_r2468780773
##########
flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/sink/dynamic/TableMetadataCache.java:
##########
@@ -50,13 +51,19 @@ class TableMetadataCache {
private final Catalog catalog;
private final long refreshMs;
+ private final Clock cacheRefreshClock;
Review Comment:
If we update the cache in `TableMetadataCache`, do we also need to sync the cache inside `DynamicWriteResultAggregator`?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix cache refreshing in DynamicIcebergSink [iceberg]
Posted by "aiborodin (via GitHub)" <gi...@apache.org>.
aiborodin commented on PR #14406:
URL: https://github.com/apache/iceberg/pull/14406#issuecomment-3435672380
@pvary Could you please take a look at this one as well?
--
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: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org