[flink] Fix cache size in GlobalIndexAssigner

apache · Sep 7, 2023 · ed3b5db · ed3b5db
1 parent c706b56
commit ed3b5db
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 3 deletions.
diff --git a/...on-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssigner.java b/...on-flink-common/src/main/java/org/apache/paimon/flink/sink/index/GlobalIndexAssigner.java
@@ -23,6 +23,7 @@
 import org.apache.paimon.data.BinaryRow;
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.data.serializer.RowCompactedSerializer;
+import org.apache.paimon.flink.RocksDBOptions;
 import org.apache.paimon.flink.lookup.RocksDBStateFactory;
 import org.apache.paimon.flink.lookup.RocksDBValueState;
 import org.apache.paimon.options.Options;
@@ -106,14 +107,13 @@ public void open(File tmpDir, int numAssigners, int assignId, BiConsumer<T, Inte
         Options options = coreOptions.toConfiguration();
         this.path = new File(tmpDir, "lookup-" + UUID.randomUUID());
         this.stateFactory = new RocksDBStateFactory(path.toString(), options);
-        long cacheSize = options.get(CoreOptions.LOOKUP_CACHE_MAX_MEMORY_SIZE).getBytes();
         RowType keyType = table.schema().logicalTrimmedPrimaryKeysType();
         this.keyIndex =
                 stateFactory.valueState(
                         "keyIndex",
                         new RowCompactedSerializer(keyType),
                         new PositiveIntIntSerializer(),
-                        cacheSize);
+                        options.get(RocksDBOptions.LOOKUP_CACHE_ROWS));
 
         this.partMapping = new IDMapping<>(BinaryRow::copy);
         this.bucketAssigner = new BucketAssigner();

diff --git a/paimon-flink/paimon-flink-common/src/test/resources/log4j2-test.properties b/paimon-flink/paimon-flink-common/src/test/resources/log4j2-test.properties
@@ -18,7 +18,7 @@
 
 # Set root logger level to OFF to not flood build logs
 # set manually to INFO for debugging purposes
-rootLogger.level = OFF
+rootLogger.level = INFO
 rootLogger.appenderRef.test.ref = TestLogger
 
 appender.testlogger.name = TestLogger