neuyilan commented on code in PR #5674:
URL: https://github.com/apache/paimon/pull/5674#discussion_r2137061475
##########
docs/layouts/shortcodes/generated/core_configuration.html:
##########
@@ -176,6 +176,12 @@
<td>Double</td>
<td>Ratio of the deleted rows in a data file to be forced
compacted for append-only table.</td>
</tr>
+ <tr>
+ <td><h5>compaction.external</h5></td>
+ <td style="word-wrap: break-word;">false</td>
+ <td>Boolean</td>
+ <td>Whether to execute full compaction to external paths.</td>
+ </tr>
Review Comment:
Perhaps we should change the name, such as
`compaction.force-pick-all-files`. Because the current implementation only
selects all files that meet the conditions and then rewrites them. Just with
the external path configurations, we can write the rewritten file to the
external path.
how do you think @JingsongLi
##########
paimon-e2e-tests/src/test/java/org/apache/paimon/tests/FlinkProceduresE2eTest.java:
##########
@@ -121,4 +121,66 @@ public void testCompact() throws Exception {
// check that second part of test data are compacted
checkResult("20221205, 1, 101", "20221206, 1, 101");
}
+
+ @Test
+ public void testExternalCompact() throws Exception {
+ // create hive catalog to test catalog loading
+ String warehouse = HDFS_ROOT + "/" + UUID.randomUUID() + ".warehouse";
+ String externalPath = HDFS_ROOT + "/" + UUID.randomUUID() +
".warehouse/external";
+ String catalogDdl =
+ String.format(
+ "CREATE CATALOG ts_catalog WITH (\n"
+ + " 'type' = 'paimon',\n"
+ + " 'warehouse' = '%s',\n"
+ + " 'metastore' = 'hive',\n"
+ + " 'uri' =
'thrift://hive-metastore:9083'\n"
+ + ");",
+ warehouse);
+ String useCatalogCmd = "USE CATALOG ts_catalog;";
+
+ String tableDdl =
+ "CREATE TABLE IF NOT EXISTS ts_table (\n"
+ + " dt STRING,\n"
+ + " k INT,\n"
+ + " v INT,\n"
+ + " PRIMARY KEY (dt, k) NOT ENFORCED\n"
+ + ") PARTITIONED BY (dt) WITH (\n"
+ + " 'data-file.external-paths' = '"
+ + externalPath
+ + "',\n"
+ + " 'bucket' = '1',\n"
+ + " 'write-only' = 'true'\n"
+ + ");";
+
+ String insert1 =
+ "INSERT INTO ts_table VALUES ('2023-01-13', 0, 0),
('2023-01-14', 0, 0), ('2023-01-13', 0, 0), "
+ + "('2023-01-15', 0, 1), ('2023-01-15', 0, 1),
('2023-01-15', 0, 1), "
+ + "('2023-01-16', 1, 0), ('2023-01-17', 1, 0),
('2023-01-18', 1, 0), "
+ + "('2023-01-19', 1, 1), ('2023-01-20', 1, 1),
('2023-01-21', 1, 1);";
+
+ String insert2 =
+ "INSERT INTO ts_table VALUES ('2023-01-13', 0, 1),
('2023-01-14', 0, 1), ('2023-01-13', 0, 1), "
+ + "('2023-01-15', 0, 2), ('2023-01-15', 0, 2),
('2023-01-15', 0, 2), "
+ + "('2023-01-16', 1, 3), ('2023-01-17', 1, 3),
('2023-01-18', 1, 3), "
+ + "('2023-01-16', 2, 3), ('2023-01-17', 2, 3),
('2023-01-18', 2, 3), ";
+
+ runBatchSql(
+ "SET 'table.dml-sync' = 'true';\n" + insert1, catalogDdl,
useCatalogCmd, tableDdl);
+ runBatchSql(
+ "SET 'table.dml-sync' = 'true';\n" + insert2, catalogDdl,
useCatalogCmd, tableDdl);
+
+ Thread.sleep(5000);
+
+ // execute external compact procedure
+ String callStatement;
+ if (System.getProperty("test.flink.main.version").compareTo("1.18") ==
0) {
+ callStatement = "CALL sys.compact('default.ts_table');";
+ } else {
+ callStatement =
+ "CALL sys.compact(table => 'default.ts_table',
partition_idle_time =>'1s',"
+ + " options =>
'compaction.external=true,data-file.external-paths.strategy=specific-fs,data-file.external-paths.specific-fs=hdfs');";
+ }
+
+ runStreamingSql(callStatement, catalogDdl, useCatalogCmd);
+ }
Review Comment:
We need to check the results of comapct, and after compacting, all new
snapshot files will be in the external path specified, may be we can verify by
obtaining the files system table of the current file.
--
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]
