[core] disable streaming read of ReadOptimizedTable which has primary…

…-key. (#4192)
apache · Sep 17, 2024 · 4e457ba · 4e457ba
1 parent 0188935
commit 4e457ba
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 0 deletions.
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/ReadOptimizedTable.java
@@ -137,6 +137,10 @@ public DataTableBatchScan newScan() {
 
     @Override
     public StreamDataTableScan newStreamScan() {
+        if (wrapped.schema().primaryKeys().size() > 0) {
+            throw new UnsupportedOperationException(
+                    "Unsupported streaming scan for read optimized table");
+        }
         return new DataTableStreamScan(
                 coreOptions(),
                 newSnapshotReader(),

diff --git a/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/PrimaryKeyFileStoreTableTest.java
@@ -1380,6 +1380,34 @@ public void testReadOptimizedTable() throws Exception {
         commit.close();
     }
 
+    @Test
+    public void testStreamingReadOptimizedTable() throws Exception {
+        FileStoreTable table =
+                createFileStoreTable(options -> options.set(TARGET_FILE_SIZE, new MemorySize(1)));
+        StreamTableWrite write = table.newWrite(commitUser);
+        StreamTableCommit commit = table.newCommit(commitUser);
+
+        write.write(rowDataWithKind(RowKind.INSERT, 1, 10, 100L));
+        write.compact(binaryRow(1), 0, true);
+        commit.commit(0, write.prepareCommit(true, 0));
+
+        ReadOptimizedTable roTable = new ReadOptimizedTable(table);
+        Function<InternalRow, String> rowDataToString =
+                row ->
+                        internalRowToString(
+                                row,
+                                DataTypes.ROW(
+                                        DataTypes.INT(), DataTypes.INT(), DataTypes.BIGINT()));
+
+        TableRead read = roTable.newRead();
+        List<String> result = getResult(read, roTable.newScan().plan().splits(), rowDataToString);
+        assertThat(result).containsExactlyInAnyOrder("+I[1, 10, 100]");
+
+        assertThatThrownBy(roTable::newStreamScan)
+                .isInstanceOf(UnsupportedOperationException.class)
+                .hasMessage("Unsupported streaming scan for read optimized table");
+    }
+
     @Test
     public void testReadDeletionVectorTable() throws Exception {
         FileStoreTable table =