This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e905dae002 [flink] Report compact_database with no matching tables in
combined mode (#10630)
e905dae002 is described below
commit e905dae0025b914da94c3c7bbc830294ff139fa0
Author: p-eye <[email protected]>
AuthorDate: Sun Oct 18 17:48:38 2026 +0900
[flink] Report compact_database with no matching tables in combined mode
(#10630)
---
docs/docs/flink/procedures/compaction.md | 2 +-
docs/docs/maintenance/dedicated-compaction.mdx | 2 +-
.../paimon/flink/action/CompactDatabaseAction.java | 11 +++---
.../paimon/flink/compact/MultiTableScanBase.java | 18 ++++++++--
.../flink/action/CompactDatabaseActionITCase.java | 39 ++++++++++++++++++++++
5 files changed, 64 insertions(+), 8 deletions(-)
diff --git a/docs/docs/flink/procedures/compaction.md
b/docs/docs/flink/procedures/compaction.md
index d2a3b889f9..28cc5c81b3 100644
--- a/docs/docs/flink/procedures/compaction.md
+++ b/docs/docs/flink/procedures/compaction.md
@@ -128,7 +128,7 @@ To compact databases. Arguments:
- `includingDatabases`: to specify databases. You can use regular expression.
-- `mode`: compact mode. "divided" (default): start a sink for each table,
detecting the new table requires restarting the job; "combined": start a single
combined sink for all tables, the new table will be automatically detected.
+- `mode`: compact mode. "divided" (default): start a sink for each table,
detecting the new table requires restarting the job; "combined": start a single
combined sink for all tables, the new table will be automatically detected. If
no table matches, a batch job fails, while a streaming job logs a warning and
keeps waiting for matching tables.
- `includingTables`: to specify tables. The format is
`databaseName.tableName`, and the whole name must match. You can use regular
expression, for example `.*\.table_.*`.
diff --git a/docs/docs/maintenance/dedicated-compaction.mdx
b/docs/docs/maintenance/dedicated-compaction.mdx
index 6632b63050..d3a2ed3824 100644
--- a/docs/docs/maintenance/dedicated-compaction.mdx
+++ b/docs/docs/maintenance/dedicated-compaction.mdx
@@ -250,7 +250,7 @@ CALL sys.compact_database(
* `--excluding_tables` is used to specify which source tables are not to be
compacted. The usage is same as "--including_tables". "--excluding_tables" has
higher priority than "--including_tables" if you specified both.
* `--mode` is used to specify compaction mode. Possible values:
* "divided" (the default mode if you haven't specified one): start a sink
for each table, the compaction of the new table requires restarting the job.
- * "combined": start a single combined sink for all tables, the new table
will be automatically compacted.
+ * "combined": start a single combined sink for all tables, the new table
will be automatically compacted. If no table matches, a batch job fails, while
a streaming job logs a warning and keeps waiting for matching tables.
* `--catalog_conf` is the configuration for Paimon catalog. Each configuration
should be specified in the format `key=value`. See [here](./configurations) for
a complete list of catalog configurations.
* `--table_conf` is the configuration for compaction. Each configuration
should be specified in the format `key=value`. Pivotal configuration is listed
below:
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
index 47323c68dc..47311dcc2d 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/action/CompactDatabaseAction.java
@@ -63,6 +63,10 @@ import static
org.apache.paimon.flink.sink.FlinkStreamPartitioner.rebalance;
public class CompactDatabaseAction extends ActionBase {
private static final Logger LOG =
LoggerFactory.getLogger(CompactDatabaseAction.class);
+ private static final String NO_TABLES_MESSAGE =
+ "no tables to be compacted. possible cause is that there are no
tables detected after pattern matching. "
+ + "Table patterns are matched against
'databaseName.tableName'.";
+
private Pattern includingPattern = Pattern.compile(".*");
@Nullable private Pattern excludingPattern;
private Pattern databasePattern = Pattern.compile(".*");
@@ -187,9 +191,7 @@ public class CompactDatabaseAction extends ActionBase {
private void buildForDividedMode() {
collectTables();
- Preconditions.checkState(
- !tableMap.isEmpty(),
- "no tables to be compacted. possible cause is that there are
no tables detected after pattern matching");
+ Preconditions.checkState(!tableMap.isEmpty(), NO_TABLES_MESSAGE);
for (Map.Entry<String, FileStoreTable> entry : tableMap.entrySet()) {
FileStoreTable fileStoreTable = entry.getValue();
@@ -213,6 +215,7 @@ public class CompactDatabaseAction extends ActionBase {
Pattern combinedIncludingPattern = includingPattern;
if (!isStreaming) {
collectTables();
+ Preconditions.checkState(!tableMap.isEmpty(), NO_TABLES_MESSAGE);
List<String> combinedTables = new ArrayList<>();
for (Map.Entry<String, FileStoreTable> entry :
tableMap.entrySet()) {
if (entry.getValue().coreOptions().dataEvolutionEnabled()) {
@@ -221,7 +224,7 @@ public class CompactDatabaseAction extends ActionBase {
combinedTables.add(Pattern.quote(entry.getKey()));
}
}
- if (combinedTables.isEmpty() && !tableMap.isEmpty()) {
+ if (combinedTables.isEmpty()) {
return;
}
// Batch table selection is fixed when building the job.
Data-evolution tables
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiTableScanBase.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiTableScanBase.java
index 67b0925a4f..8ed664a493 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiTableScanBase.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/compact/MultiTableScanBase.java
@@ -81,14 +81,19 @@ public abstract class MultiTableScanBase<T> implements
AutoCloseable {
protected void updateTableMap()
throws Catalog.DatabaseNotExistException,
Catalog.TableNotExistException {
List<String> databases = catalog.listDatabases();
+ // Counted across bucket types: the two combined sources each see all
matched tables.
+ int matchedTables = 0;
for (String databaseName : databases) {
if (databasePattern.matcher(databaseName).matches()) {
List<String> tables = catalog.listTables(databaseName);
for (String tableName : tables) {
Identifier identifier = Identifier.create(databaseName,
tableName);
- if (shouldCompactTable(identifier, includingPattern,
excludingPattern)
- && (!checkTableScanned(identifier))) {
+ if (!shouldCompactTable(identifier, includingPattern,
excludingPattern)) {
+ continue;
+ }
+ matchedTables++;
+ if (!checkTableScanned(identifier)) {
Table table = catalog.getTable(identifier);
if (!(table instanceof FileStoreTable)) {
LOG.error(
@@ -104,6 +109,15 @@ public abstract class MultiTableScanBase<T> implements
AutoCloseable {
}
}
}
+
+ if (matchedTables == 0) {
+ LOG.warn(
+ "No tables match database pattern '{}', including pattern
'{}' and excluding pattern '{}'. "
+ + "Table patterns are matched against
'databaseName.tableName'.",
+ databasePattern,
+ includingPattern,
+ excludingPattern);
+ }
}
public ScanResult scanTable(ReaderOutput<T> ctx)
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
index b1534c9fae..5528526470 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/action/CompactDatabaseActionITCase.java
@@ -457,6 +457,45 @@ public class CompactDatabaseActionITCase extends
CompactActionITCaseBase {
.hasMessageContaining("Data evolution table compact only
supports batch mode");
}
+ @ParameterizedTest(name = "mode = {0}")
+ @ValueSource(strings = {"divided", "combined"})
+ public void testNoTablesMatchedInBatchMode(String mode) throws Exception {
+ createTable(
+ "db1",
+ "t1",
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ Collections.singletonMap("bucket", "1"));
+ CompactDatabaseAction action =
+ new
CompactDatabaseAction(Collections.singletonMap("warehouse", warehouse))
+ .includingTables("t1")
+ .withDatabaseCompactMode(mode);
+ action.withStreamExecutionEnvironment(
+ streamExecutionEnvironmentBuilder().batchMode().build());
+ assertThatThrownBy(action::build)
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessageContaining("no tables to be compacted")
+ .hasMessageContaining("'databaseName.tableName'");
+ }
+
+ @Test
+ public void testNoTablesMatchedInCombinedStreamingMode() throws Exception {
+ createTable(
+ "db1",
+ "t1",
+ Arrays.asList("dt", "hh"),
+ Arrays.asList("dt", "hh", "k"),
+ Collections.singletonMap("bucket", "1"));
+ CompactDatabaseAction action =
+ new
CompactDatabaseAction(Collections.singletonMap("warehouse", warehouse))
+ .includingTables("t1")
+ .withDatabaseCompactMode("combined");
+ action.withStreamExecutionEnvironment(
+ streamExecutionEnvironmentBuilder().streamingMode().build());
+ // Tables matching the patterns may be created after the streaming job
starts.
+ action.build();
+ }
+
private void writeDataEvolutionRecords(FileStoreTable table) throws
Exception {
BatchWriteBuilder builder = table.newBatchWriteBuilder();
try (BatchTableWrite write = builder.newWrite();
