Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[core][bug]Fix partition table get index RowType error #3885

Merged
merged 4 commits into from
Aug 6, 2024
Merged

[core][bug]Fix partition table get index RowType error #3885

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
6480ab7
[core][bug]Fix partition table get index RowType error
hawk9821 Aug 3, 2024
a838be6
[core][bug]Fix partition table get index RowType error
hawk9821 Aug 4, 2024
d0e0c76
[core][bug]Fix partition table get index RowType error
hawk9821 Aug 5, 2024
fa40316
[core][bug]Fix partition table get index RowType error
hawk9821 Aug 5, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions paimon-core/src/main/java/org/apache/paimon/crosspartition/IndexBootstrap.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,6 +142,7 @@ public static RowType bootstrapType(TableSchema schema) {
new ArrayList<>(
schema.projectedLogicalRowType(
Stream.concat(primaryKeys.stream(), partitionKeys.stream())
.distinct()
.collect(Collectors.toList()))
.getFields());
bootstrapFields.add(
Expand Down
44 changes: 30 additions & 14 deletions paimon-core/src/test/java/org/apache/paimon/crosspartition/IndexBootstrapTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,10 +27,12 @@
import org.apache.paimon.manifest.FileSource;
import org.apache.paimon.options.Options;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.TableTestBase;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Pair;

import org.junit.jupiter.api.Test;
Expand All @@ -43,6 +45,7 @@
import java.util.List;
import java.util.function.Consumer;

import static org.apache.paimon.crosspartition.IndexBootstrap.BUCKET_FIELD;
import static org.apache.paimon.crosspartition.IndexBootstrap.filterSplit;
import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
Expand All @@ -53,20 +56,7 @@ public class IndexBootstrapTest extends TableTestBase {

@Test
public void testBoostrap() throws Exception {
Identifier identifier = identifier("T");
Options options = new Options();
options.set(CoreOptions.BUCKET, -1);
Schema schema =
Schema.newBuilder()
.column("pt", DataTypes.INT())
.column("col", DataTypes.INT())
.column("pk", DataTypes.INT())
.primaryKey("pk")
.partitionKeys("pt")
.options(options.toMap())
.build();
catalog.createTable(identifier, schema, true);
Table table = catalog.getTable(identifier);
Table table = createTable();

write(
table,
Expand Down Expand Up @@ -106,6 +96,32 @@ public void testBoostrap() throws Exception {
Thread.sleep(1000);
}

private Table createTable() throws Exception {
Identifier identifier = identifier("T");
Options options = new Options();
options.set(CoreOptions.BUCKET, -1);
Schema schema =
Schema.newBuilder()
.column("pt", DataTypes.INT())
.column("col", DataTypes.INT())
.column("pk", DataTypes.INT())
.primaryKey("pk")
.partitionKeys("pt")
.options(options.toMap())
.build();
catalog.createTable(identifier, schema, true);
return catalog.getTable(identifier);
}

@Test
public void testBootstrapType() throws Exception {
FileStoreTable table = (FileStoreTable) createTable();
RowType indexRowType = IndexBootstrap.bootstrapType(table.schema());
assertThat(indexRowType.getFieldNames()).contains(BUCKET_FIELD);
assertThat(indexRowType.getFieldIndex(BUCKET_FIELD))
.isEqualTo(2); // the last field is bucket, which is not in table schema
}

@Test
public void testFilterSplit() {
assertThat(filterSplit(newSplit(newFile(100), newFile(200)), 50, 230)).isTrue();
Expand Down
32 changes: 32 additions & 0 deletions ...on-flink-common/src/test/java/org/apache/paimon/flink/GlobalDynamicBucketTableITCase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,21 @@

package org.apache.paimon.flink;

import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.CatalogFactory;
import org.apache.paimon.catalog.Identifier;
import org.apache.paimon.crosspartition.IndexBootstrap;
import org.apache.paimon.fs.Path;
import org.apache.paimon.table.FileStoreTable;

import org.apache.flink.types.Row;
import org.junit.jupiter.api.Test;

import java.util.Arrays;
import java.util.List;

import static org.apache.paimon.crosspartition.IndexBootstrap.BUCKET_FIELD;
import static org.assertj.core.api.Assertions.assertThat;

/** ITCase for batch file store. */
Expand Down Expand Up @@ -161,4 +170,27 @@ public void testLargeRecords() {
sql("insert into large_t select * from src");
assertThat(sql("select k, count(*) from large_t group by k having count(*) > 1")).isEmpty();
}

@Test
public void testBootstrapType() throws Exception {
Catalog catalog = CatalogFactory.createCatalog(CatalogContext.create(new Path(path)));
FileStoreTable t = (FileStoreTable) catalog.getTable(Identifier.create("default", "T"));
FileStoreTable partialUpdateT =
(FileStoreTable) catalog.getTable(Identifier.create("default", "partial_update_t"));
FileStoreTable firstRowT =
(FileStoreTable) catalog.getTable(Identifier.create("default", "first_row_t"));
assertThat(IndexBootstrap.bootstrapType(t.schema()).getFieldNames()).contains(BUCKET_FIELD);
assertThat(IndexBootstrap.bootstrapType(partialUpdateT.schema()).getFieldNames())
.contains(BUCKET_FIELD);
assertThat(IndexBootstrap.bootstrapType(firstRowT.schema()).getFieldNames())
.contains(BUCKET_FIELD);
assertThat(IndexBootstrap.bootstrapType(t.schema()).getFieldIndex(BUCKET_FIELD))
.isEqualTo(2);
assertThat(
IndexBootstrap.bootstrapType(partialUpdateT.schema())
.getFieldIndex(BUCKET_FIELD))
.isEqualTo(2);
assertThat(IndexBootstrap.bootstrapType(firstRowT.schema()).getFieldIndex(BUCKET_FIELD))
.isEqualTo(2);
}
}
Loading