Skip to content

Commit

Permalink
[core] supports using dynamic parameters to query the system table of…
Browse files Browse the repository at this point in the history 
… a specified branch. (#4527)
  • Loading branch information
@liming30
liming30 authored Nov 14, 2024
1 parent c7dfcfa commit c95c3e6
Show file tree
Hide file tree
Showing 8 changed files with 97 additions and 75 deletions.
18 changes: 7 additions & 11 deletions paimon-core/src/main/java/org/apache/paimon/table/system/AggregationFieldsTable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -80,17 +80,13 @@ public class AggregationFieldsTable implements ReadonlyTable {
private final Path location;
private final String branch;

public AggregationFieldsTable(FileStoreTable dataTable) {
this(
dataTable.fileIO(),
dataTable.location(),
CoreOptions.branch(dataTable.schema().options()));
}
private final FileStoreTable dataTable;

public AggregationFieldsTable(FileIO fileIO, Path location, String branchName) {
this.fileIO = fileIO;
this.location = location;
this.branch = branchName;
public AggregationFieldsTable(FileStoreTable dataTable) {
this.fileIO = dataTable.fileIO();
this.location = dataTable.location();
this.branch = CoreOptions.branch(dataTable.schema().options());
this.dataTable = dataTable;
}

@Override
Expand Down Expand Up @@ -120,7 +116,7 @@ public InnerTableRead newRead() {

@Override
public Table copy(Map<String, String> dynamicOptions) {
return new AggregationFieldsTable(fileIO, location, branch);
return new AggregationFieldsTable(dataTable.copy(dynamicOptions));
}

private class SchemasScan extends ReadOnceTableScan {
Expand Down
13 changes: 6 additions & 7 deletions paimon-core/src/main/java/org/apache/paimon/table/system/BranchesTable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -81,13 +81,12 @@ public class BranchesTable implements ReadonlyTable {
private final FileIO fileIO;
private final Path location;

public BranchesTable(FileStoreTable dataTable) {
this(dataTable.fileIO(), dataTable.location());
}
private final FileStoreTable dataTable;

public BranchesTable(FileIO fileIO, Path location) {
this.fileIO = fileIO;
this.location = location;
public BranchesTable(FileStoreTable dataTable) {
this.fileIO = dataTable.fileIO();
this.location = dataTable.location();
this.dataTable = dataTable;
}

@Override
Expand Down Expand Up @@ -117,7 +116,7 @@ public InnerTableRead newRead() {

@Override
public Table copy(Map<String, String> dynamicOptions) {
return new BranchesTable(fileIO, location);
return new BranchesTable(dataTable.copy(dynamicOptions));
}

private class BranchesScan extends ReadOnceTableScan {
Expand Down
18 changes: 7 additions & 11 deletions paimon-core/src/main/java/org/apache/paimon/table/system/ConsumersTable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -74,17 +74,13 @@ public class ConsumersTable implements ReadonlyTable {
private final Path location;
private final String branch;

public ConsumersTable(FileStoreTable dataTable) {
this(
dataTable.fileIO(),
dataTable.location(),
CoreOptions.branch(dataTable.schema().options()));
}
private final FileStoreTable dataTable;

public ConsumersTable(FileIO fileIO, Path location, String branchName) {
this.fileIO = fileIO;
this.location = location;
this.branch = branchName;
public ConsumersTable(FileStoreTable dataTable) {
this.fileIO = dataTable.fileIO();
this.location = dataTable.location();
this.branch = CoreOptions.branch(dataTable.schema().options());
this.dataTable = dataTable;
}

@Override
Expand Down Expand Up @@ -114,7 +110,7 @@ public InnerTableRead newRead() {

@Override
public Table copy(Map<String, String> dynamicOptions) {
return new ConsumersTable(fileIO, location, branch);
return new ConsumersTable(dataTable.copy(dynamicOptions));
}

private class ConsumersScan extends ReadOnceTableScan {
Expand Down
18 changes: 7 additions & 11 deletions paimon-core/src/main/java/org/apache/paimon/table/system/OptionsTable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -72,17 +72,13 @@ public class OptionsTable implements ReadonlyTable {
private final Path location;
private final String branch;

public OptionsTable(FileStoreTable dataTable) {
this(
dataTable.fileIO(),
dataTable.location(),
CoreOptions.branch(dataTable.schema().options()));
}
private final FileStoreTable dataTable;

public OptionsTable(FileIO fileIO, Path location, String branchName) {
this.fileIO = fileIO;
this.location = location;
this.branch = branchName;
public OptionsTable(FileStoreTable dataTable) {
this.fileIO = dataTable.fileIO();
this.location = dataTable.location();
this.branch = CoreOptions.branch(dataTable.schema().options());
this.dataTable = dataTable;
}

@Override
Expand Down Expand Up @@ -112,7 +108,7 @@ public InnerTableRead newRead() {

@Override
public Table copy(Map<String, String> dynamicOptions) {
return new OptionsTable(fileIO, location, branch);
return new OptionsTable(dataTable.copy(dynamicOptions));
}

private class OptionsScan extends ReadOnceTableScan {
Expand Down
18 changes: 7 additions & 11 deletions paimon-core/src/main/java/org/apache/paimon/table/system/SchemasTable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -102,17 +102,13 @@ public class SchemasTable implements ReadonlyTable {
private final Path location;
private final String branch;

public SchemasTable(FileStoreTable dataTable) {
this(
dataTable.fileIO(),
dataTable.location(),
CoreOptions.branch(dataTable.schema().options()));
}
private final FileStoreTable dataTable;

public SchemasTable(FileIO fileIO, Path location, String branchName) {
this.fileIO = fileIO;
this.location = location;
this.branch = branchName;
public SchemasTable(FileStoreTable dataTable) {
this.fileIO = dataTable.fileIO();
this.location = dataTable.location();
this.branch = CoreOptions.branch(dataTable.schema().options());
this.dataTable = dataTable;
}

@Override
Expand Down Expand Up @@ -142,7 +138,7 @@ public InnerTableRead newRead() {

@Override
public Table copy(Map<String, String> dynamicOptions) {
return new SchemasTable(fileIO, location, branch);
return new SchemasTable(dataTable.copy(dynamicOptions));
}

private class SchemasScan extends ReadOnceTableScan {
Expand Down
17 changes: 4 additions & 13 deletions paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -116,19 +116,10 @@ public class SnapshotsTable implements ReadonlyTable {
private final FileStoreTable dataTable;

public SnapshotsTable(FileStoreTable dataTable) {
this(
dataTable.fileIO(),
dataTable.location(),
dataTable,
CoreOptions.branch(dataTable.schema().options()));
}

public SnapshotsTable(
FileIO fileIO, Path location, FileStoreTable dataTable, String branchName) {
this.fileIO = fileIO;
this.location = location;
this.fileIO = dataTable.fileIO();
this.location = dataTable.location();
this.branch = CoreOptions.branch(dataTable.schema().options());
this.dataTable = dataTable;
this.branch = branchName;
}

@Override
Expand Down Expand Up @@ -158,7 +149,7 @@ public InnerTableRead newRead() {

@Override
public Table copy(Map<String, String> dynamicOptions) {
return new SnapshotsTable(fileIO, location, dataTable.copy(dynamicOptions), branch);
return new SnapshotsTable(dataTable.copy(dynamicOptions));
}

private class SnapshotsScan extends ReadOnceTableScan {
Expand Down
18 changes: 7 additions & 11 deletions paimon-core/src/main/java/org/apache/paimon/table/system/TagsTable.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -95,17 +95,13 @@ public class TagsTable implements ReadonlyTable {
private final Path location;
private final String branch;

public TagsTable(FileStoreTable dataTable) {
this(
dataTable.fileIO(),
dataTable.location(),
CoreOptions.branch(dataTable.schema().options()));
}
private final FileStoreTable dataTable;

public TagsTable(FileIO fileIO, Path location, String branchName) {
this.fileIO = fileIO;
this.location = location;
this.branch = branchName;
public TagsTable(FileStoreTable dataTable) {
this.fileIO = dataTable.fileIO();
this.location = dataTable.location();
this.branch = CoreOptions.branch(dataTable.schema().options());
this.dataTable = dataTable;
}

@Override
Expand Down Expand Up @@ -135,7 +131,7 @@ public InnerTableRead newRead() {

@Override
public Table copy(Map<String, String> dynamicOptions) {
return new TagsTable(fileIO, location, branch);
return new TagsTable(dataTable.copy(dynamicOptions));
}

private class TagsScan extends ReadOnceTableScan {
Expand Down
52 changes: 52 additions & 0 deletions paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BranchSqlITCase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -347,6 +347,11 @@ public void testBranchOptionsTable() throws Exception {
"+I[bucket, 2]",
"+I[snapshot.time-retained, 1 h]",
"+I[scan.infer-parallelism, false]");
assertThat(collectResult("SELECT * FROM t$options /*+ OPTIONS('branch'='test') */"))
.containsExactlyInAnyOrder(
"+I[bucket, 2]",
"+I[snapshot.time-retained, 1 h]",
"+I[scan.infer-parallelism, false]");
}

@Test
Expand All @@ -360,6 +365,10 @@ public void testBranchSchemasTable() throws Exception {
sql("ALTER TABLE t$branch_b1 SET ('snapshot.time-retained' = '5 h')");
assertThat(collectResult("SELECT schema_id FROM t$branch_b1$schemas order by schema_id"))
.containsExactlyInAnyOrder("+I[0]", "+I[1]");
assertThat(
collectResult(
"SELECT schema_id FROM t$schemas /*+ OPTIONS('branch'='b1') */ order by schema_id"))
.containsExactlyInAnyOrder("+I[0]", "+I[1]");
}

@Test
Expand All @@ -373,6 +382,8 @@ public void testBranchAuditLogTable() throws Exception {
sql("INSERT INTO t$branch_b1 VALUES (3, 4)");
assertThat(collectResult("SELECT * FROM t$branch_b1$audit_log"))
.containsExactlyInAnyOrder("+I[+I, 3, 4]");
assertThat(collectResult("SELECT * FROM t$audit_log /*+ OPTIONS('branch'='b1') */"))
.containsExactlyInAnyOrder("+I[+I, 3, 4]");
}

@Test
Expand All @@ -385,6 +396,8 @@ public void testBranchReadOptimizedTable() throws Exception {
sql("INSERT INTO t$branch_b1 VALUES (3, 4)");
assertThat(collectResult("SELECT * FROM t$branch_b1$ro"))
.containsExactlyInAnyOrder("+I[3, 4]");
assertThat(collectResult("SELECT * FROM t$ro /*+ OPTIONS('branch'='b1') */"))
.containsExactlyInAnyOrder("+I[3, 4]");
}

@Test
Expand All @@ -400,6 +413,10 @@ public void testBranchFilesTable() throws Exception {
.containsExactlyInAnyOrder("+I[{a=1, b=2}]");
assertThat(collectResult("SELECT min_value_stats FROM t$branch_b1$files"))
.containsExactlyInAnyOrder("+I[{a=3, b=4}]", "+I[{a=5, b=6}]");
assertThat(
collectResult(
"SELECT min_value_stats FROM t$files /*+ OPTIONS('branch'='b1') */"))
.containsExactlyInAnyOrder("+I[{a=3, b=4}]", "+I[{a=5, b=6}]");
}

@Test
Expand All @@ -416,6 +433,10 @@ public void testBranchTagsTable() throws Exception {
.containsExactlyInAnyOrder("+I[tag1, 1, 1]");
assertThat(collectResult("SELECT tag_name,snapshot_id,record_count FROM t$branch_b1$tags"))
.containsExactlyInAnyOrder("+I[tag1, 1, 1]", "+I[tag2, 2, 2]");
assertThat(
collectResult(
"SELECT tag_name,snapshot_id,record_count FROM t$tags /*+ OPTIONS('branch'='b1') */"))
.containsExactlyInAnyOrder("+I[tag1, 1, 1]", "+I[tag2, 2, 2]");
}

@Test
Expand All @@ -435,6 +456,8 @@ public void testBranchConsumersTable() throws Exception {
assertThat(collectResult("SELECT * FROM t$consumers")).isEmpty();
assertThat(collectResult("SELECT * FROM t$branch_b1$consumers"))
.containsExactlyInAnyOrder("+I[id1, 2]");
assertThat(collectResult("SELECT * FROM t$consumers /*+ OPTIONS('branch'='b1') */"))
.containsExactlyInAnyOrder("+I[id1, 2]");
}

@Test
Expand All @@ -458,6 +481,31 @@ public void testBranchManifestsTable() {
.isTrue();
assertThat((long) row.getField(2)).isGreaterThan(0L);
});
List<Row> dynamicOptionRes =
sql(
"SELECT schema_id, file_name, file_size FROM t$manifests /*+ OPTIONS('branch'='b1') */");
assertThat(dynamicOptionRes).containsExactlyInAnyOrderElementsOf(res);
}

@Test
public void testBranchSnapshotsTable() throws Exception {
sql("CREATE TABLE t (a INT, b INT)");
sql("INSERT INTO t VALUES (1, 2)");

sql("CALL sys.create_branch('default.t', 'b1')");
sql("INSERT INTO t$branch_b1 VALUES (3, 4)");
sql("INSERT INTO t$branch_b1 VALUES (5, 6)");

assertThat(collectResult("SELECT snapshot_id, schema_id, commit_kind FROM t$snapshots"))
.containsExactlyInAnyOrder("+I[1, 0, APPEND]");
assertThat(
collectResult(
"SELECT snapshot_id, schema_id, commit_kind FROM t$branch_b1$snapshots"))
.containsExactlyInAnyOrder("+I[1, 0, APPEND]", "+I[2, 0, APPEND]");
assertThat(
collectResult(
"SELECT snapshot_id, schema_id, commit_kind FROM t$snapshots /*+ OPTIONS('branch'='b1') */"))
.containsExactlyInAnyOrder("+I[1, 0, APPEND]", "+I[2, 0, APPEND]");
}

@Test
Expand All @@ -479,6 +527,10 @@ public void testBranchPartitionsTable() throws Exception {
collectResult(
"SELECT `partition`, record_count, file_count FROM t$branch_b1$partitions"))
.containsExactlyInAnyOrder("+I[[1], 2, 2]", "+I[[2], 3, 2]");
assertThat(
collectResult(
"SELECT `partition`, record_count, file_count FROM t$partitions /*+ OPTIONS('branch'='b1') */"))
.containsExactlyInAnyOrder("+I[[1], 2, 2]", "+I[[2], 3, 2]");
}

@Test
Expand Down

0 comments on commit c95c3e6

Please sign in to comment.