This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 5c2f1b4 [FLINK-27540] Let FileStoreSource accept pre-planned manifest
entries
5c2f1b4 is described below
commit 5c2f1b4ac8c503e9d4212ee6d65b63dd355f02c9
Author: Jane Chan <[email protected]>
AuthorDate: Tue May 10 20:26:12 2022 +0800
[FLINK-27540] Let FileStoreSource accept pre-planned manifest entries
This closes #111
---
.../flink/table/store/connector/TableStore.java | 3 +-
.../store/connector/source/FileStoreSource.java | 27 +-
.../source/FileStoreSourceSplitGenerator.java | 10 +-
.../connector/source/FileStoreSourceTest.java | 292 +++++++++++++++++++++
.../flink/table/store/file/stats/FieldStats.java | 2 +-
.../store/file/utils/PartitionedManifestMeta.java | 151 +++++++++++
6 files changed, 481 insertions(+), 4 deletions(-)
diff --git
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStore.java
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStore.java
index c17a309..ae128ff 100644
---
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStore.java
+++
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/TableStore.java
@@ -320,7 +320,8 @@ public class TableStore {
continuousScanLatest,
projectedFields,
partitionPredicate,
- fieldPredicate);
+ fieldPredicate,
+ null);
}
private Source<RowData, ?, ?> buildSource() {
diff --git
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSource.java
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSource.java
index 93faca2..4eacd73 100644
---
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSource.java
+++
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSource.java
@@ -18,6 +18,7 @@
package org.apache.flink.table.store.connector.source;
+import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.Source;
import org.apache.flink.api.connector.source.SourceReader;
@@ -30,6 +31,7 @@ import org.apache.flink.table.store.file.Snapshot;
import org.apache.flink.table.store.file.operation.FileStoreRead;
import org.apache.flink.table.store.file.operation.FileStoreScan;
import org.apache.flink.table.store.file.predicate.Predicate;
+import org.apache.flink.table.store.file.utils.PartitionedManifestMeta;
import javax.annotation.Nullable;
@@ -61,6 +63,12 @@ public class FileStoreSource
@Nullable private final Predicate fieldPredicate;
+ /**
+ * The partitioned manifest meta collected at planning phase when manual
compaction is
+ * triggered.
+ */
+ @Nullable private final PartitionedManifestMeta specifiedPartManifests;
+
public FileStoreSource(
FileStore fileStore,
boolean valueCountMode,
@@ -69,7 +77,8 @@ public class FileStoreSource
boolean latestContinuous,
@Nullable int[][] projectedFields,
@Nullable Predicate partitionPredicate,
- @Nullable Predicate fieldPredicate) {
+ @Nullable Predicate fieldPredicate,
+ @Nullable PartitionedManifestMeta specifiedPartManifests) {
this.fileStore = fileStore;
this.valueCountMode = valueCountMode;
this.isContinuous = isContinuous;
@@ -78,6 +87,7 @@ public class FileStoreSource
this.projectedFields = projectedFields;
this.partitionPredicate = partitionPredicate;
this.fieldPredicate = fieldPredicate;
+ this.specifiedPartManifests = specifiedPartManifests;
}
@Override
@@ -122,6 +132,15 @@ public class FileStoreSource
SplitEnumeratorContext<FileStoreSourceSplit> context,
PendingSplitsCheckpoint checkpoint) {
FileStoreScan scan = fileStore.newScan();
+
+ if (specifiedPartManifests != null) {
+ return new StaticFileStoreSplitEnumerator(
+ context,
+ scan.snapshot(specifiedPartManifests.getSnapshotId()),
+ new FileStoreSourceSplitGenerator()
+
.createSplits(specifiedPartManifests.getManifestEntries()));
+ }
+
if (partitionPredicate != null) {
scan.withPartitionFilter(partitionPredicate);
}
@@ -182,4 +201,10 @@ public class FileStoreSource
public PendingSplitsCheckpointSerializer
getEnumeratorCheckpointSerializer() {
return new PendingSplitsCheckpointSerializer(getSplitSerializer());
}
+
+ @VisibleForTesting
+ @Nullable
+ PartitionedManifestMeta getSpecifiedPartManifests() {
+ return specifiedPartManifests;
+ }
}
diff --git
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGenerator.java
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGenerator.java
index f9b64ab..9e438e1 100644
---
a/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGenerator.java
+++
b/flink-table-store-connector/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitGenerator.java
@@ -18,9 +18,12 @@
package org.apache.flink.table.store.connector.source;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.store.file.data.DataFileMeta;
import org.apache.flink.table.store.file.operation.FileStoreScan;
import java.util.List;
+import java.util.Map;
import java.util.stream.Collectors;
/**
@@ -36,7 +39,12 @@ public class FileStoreSourceSplitGenerator {
private final char[] currentId = "0000000000".toCharArray();
public List<FileStoreSourceSplit> createSplits(FileStoreScan.Plan plan) {
- return plan.groupByPartFiles().entrySet().stream()
+ return createSplits(plan.groupByPartFiles());
+ }
+
+ public List<FileStoreSourceSplit> createSplits(
+ Map<BinaryRowData, Map<Integer, List<DataFileMeta>>> groupBy) {
+ return groupBy.entrySet().stream()
.flatMap(
pe ->
pe.getValue().entrySet().stream()
diff --git
a/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceTest.java
b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceTest.java
new file mode 100644
index 0000000..e5f6584
--- /dev/null
+++
b/flink-table-store-connector/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceTest.java
@@ -0,0 +1,292 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.store.connector.source;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
+import org.apache.flink.table.store.file.FileStore;
+import org.apache.flink.table.store.file.FileStoreImpl;
+import org.apache.flink.table.store.file.data.DataFileMeta;
+import
org.apache.flink.table.store.file.mergetree.compact.DeduplicateMergeFunction;
+import org.apache.flink.table.store.file.mergetree.compact.MergeFunction;
+import
org.apache.flink.table.store.file.mergetree.compact.ValueCountMergeFunction;
+import org.apache.flink.table.store.file.stats.FieldStats;
+import org.apache.flink.table.store.file.utils.PartitionedManifestMeta;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.DoubleType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link FileStoreSource}. */
+public class FileStoreSourceTest {
+
+ private static final RowType RECORD_TYPE =
+ RowType.of(
+ new LogicalType[] {
+ new IntType(), new VarCharType(), new DoubleType(),
new VarCharType()
+ },
+ new String[] {"k0", "k1", "v0", "v1"});
+
+ @MethodSource("parameters")
+ @ParameterizedTest
+ public void testSerDe(boolean hasPk, boolean partitioned, boolean
specified)
+ throws ClassNotFoundException, IOException {
+ FileStore fileStore = buildFileStore(hasPk, partitioned);
+ Long specifiedSnapshotId = specified ? 1L : null;
+ Map<BinaryRowData, Map<Integer, List<DataFileMeta>>>
specifiedManifestEntries =
+ specified ? buildManifestEntries(hasPk, partitioned) : null;
+ PartitionedManifestMeta partitionedManifestMeta =
+ specified
+ ? new PartitionedManifestMeta(
+ specifiedSnapshotId,
+ specifiedManifestEntries,
+ getPartitionType(partitioned).getFieldCount(),
+ getKeyType(hasPk),
+ getValueType(hasPk))
+ : null;
+ FileStoreSource source =
+ new FileStoreSource(
+ fileStore,
+ !hasPk,
+ true,
+ Duration.ofSeconds(1).toMillis(),
+ true,
+ null,
+ null,
+ null,
+ partitionedManifestMeta);
+ Object object = readObject(writeObject(source));
+ assertThat(object).isInstanceOf(FileStoreSource.class);
+ FileStoreSource deserialized = (FileStoreSource) object;
+
assertThat(deserialized.getBoundedness()).isEqualTo(source.getBoundedness());
+ if (specified) {
+ assertThat(deserialized.getSpecifiedPartManifests())
+ .isEqualTo(source.getSpecifiedPartManifests());
+ } else {
+ assertThat(deserialized.getSpecifiedPartManifests()).isNull();
+ }
+ }
+
+ private byte[] writeObject(FileStoreSource source) throws IOException {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream(4096);
+ ObjectOutputStream oos = new ObjectOutputStream(baos);
+ oos.writeObject(source);
+ oos.close();
+ return baos.toByteArray();
+ }
+
+ private Object readObject(byte[] bytes) throws IOException,
ClassNotFoundException {
+ ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
+ ObjectInputStream ois = new ObjectInputStream(bais);
+ Object object = ois.readObject();
+ ois.close();
+ return object;
+ }
+
+ public static Stream<Arguments> parameters() {
+ // hasPk, partitioned, specified
+ return Stream.of(
+ Arguments.of(true, true, false),
+ Arguments.of(true, false, false),
+ Arguments.of(false, false, false),
+ Arguments.of(false, true, false),
+ Arguments.of(true, true, true),
+ Arguments.of(true, false, true),
+ Arguments.of(false, false, true),
+ Arguments.of(false, true, true));
+ }
+
+ private static FileStore buildFileStore(boolean hasPk, boolean
partitioned) {
+ ObjectIdentifier tableIdentifier = ObjectIdentifier.of("cat", "db",
"tbl");
+ String user = "user";
+ RowType partitionType = getPartitionType(partitioned);
+ RowType keyType = getKeyType(hasPk);
+ RowType valueType = getValueType(hasPk);
+ MergeFunction mergeFunction =
+ hasPk ? new DeduplicateMergeFunction() : new
ValueCountMergeFunction();
+ return new FileStoreImpl(
+ tableIdentifier,
+ new Configuration(),
+ user,
+ partitionType,
+ keyType,
+ valueType,
+ mergeFunction);
+ }
+
+ private static Map<BinaryRowData, Map<Integer, List<DataFileMeta>>>
buildManifestEntries(
+ boolean hasPk, boolean partitioned) {
+ Map<BinaryRowData, Map<Integer, List<DataFileMeta>>> manifestEntries =
new HashMap<>();
+ Map<Integer, List<DataFileMeta>> bucketEntries = new HashMap<>();
+ int totalBuckets = new Random().nextInt(10) + 1;
+ for (int bucket = 0; bucket < totalBuckets; bucket++) {
+ List<DataFileMeta> metaList = new ArrayList<>();
+ for (Tuple2<BinaryRowData, BinaryRowData> tuple :
genMinMax(hasPk)) {
+ metaList.add(genDataFileMeta(hasPk, tuple.f0, tuple.f1));
+ }
+ bucketEntries.put(bucket, metaList);
+ }
+ genPartitionValues(partitioned)
+ .forEach(partValue -> manifestEntries.put(partValue,
bucketEntries));
+ return manifestEntries;
+ }
+
+ private static List<BinaryRowData> genPartitionValues(boolean partitioned)
{
+ RowDataSerializer partSerializer = new
RowDataSerializer(getPartitionType(partitioned));
+ if (partitioned) {
+ int partSize = new Random().nextInt(10) + 1;
+ List<BinaryRowData> partKeys = new ArrayList<>();
+ for (int i = 0; i < partSize; i++) {
+ partKeys.add(
+ partSerializer
+ .toBinaryRow(
+ GenericRowData.of(
+ StringData.fromString(
+
UUID.randomUUID().toString())))
+ .copy());
+ }
+ return partKeys;
+ }
+ return
Collections.singletonList(partSerializer.toBinaryRow(GenericRowData.of()).copy());
+ }
+
+ private static List<Tuple2<BinaryRowData, BinaryRowData>>
genMinMax(boolean hasPk) {
+ RowDataSerializer keySerializer = new
RowDataSerializer(getKeyType(hasPk));
+ int size = new Random().nextInt(20);
+ List<Tuple2<BinaryRowData, BinaryRowData>> minMaxKeys = new
ArrayList<>();
+ for (int i = 0; i < size; i++) {
+ int k0 = new Random().nextInt(1000);
+ String k1 = UUID.randomUUID().toString();
+ BinaryRowData keyMin;
+ BinaryRowData keyMax;
+ if (hasPk) {
+ keyMin =
+ keySerializer.toBinaryRow(GenericRowData.of(k0,
StringData.fromString(k1)));
+ keyMax =
+ keySerializer.toBinaryRow(
+ GenericRowData.of(k0 + 100,
StringData.fromString(k1)));
+ } else {
+ double v0 = new Random().nextDouble();
+ String v1 = UUID.randomUUID().toString();
+ keyMin =
+ keySerializer.toBinaryRow(
+ GenericRowData.of(
+ k0,
+ StringData.fromString(k1),
+ v0,
+ StringData.fromString(v1)));
+ keyMax =
+ keySerializer.toBinaryRow(
+ GenericRowData.of(
+ k0 + 100,
+ StringData.fromString(k1),
+ v0 + 1.5,
+ StringData.fromString(v1)));
+ }
+ minMaxKeys.add(Tuple2.of(keyMin, keyMax));
+ }
+ return minMaxKeys;
+ }
+
+ private static DataFileMeta genDataFileMeta(
+ boolean hasPk, BinaryRowData keyMin, BinaryRowData keyMax) {
+ long seqNumber = new Random().nextLong() + 1;
+ FieldStats k0Stats = new FieldStats(keyMin.getInt(0),
keyMax.getInt(1), 0);
+ FieldStats k1Stats = new FieldStats(keyMin.getString(1),
keyMax.getString(1), 0);
+ FieldStats v0Status =
+ new FieldStats(
+ hasPk ? null : keyMin.getDouble(2), hasPk ? null :
keyMax.getDouble(2), 0);
+ FieldStats v1Status =
+ new FieldStats(
+ hasPk ? null : keyMin.getString(3), hasPk ? null :
keyMax.getString(3), 0);
+ long count = new Random().nextLong();
+ return new DataFileMeta(
+ "data-" + UUID.randomUUID(),
+ new Random().nextInt(100),
+ new Random().nextInt(100),
+ keyMin,
+ keyMax,
+ hasPk
+ ? new FieldStats[] {k0Stats, k1Stats}
+ : new FieldStats[] {
+ k0Stats, k1Stats,
+ v0Status, v1Status
+ },
+ hasPk
+ ? new FieldStats[] {
+ k0Stats, k1Stats,
+ v0Status, v1Status
+ }
+ : new FieldStats[] {
+ new FieldStats(count, count + new
Random().nextInt(100), 0)
+ },
+ seqNumber,
+ seqNumber + new Random().nextInt(100),
+ new Random().nextInt(4));
+ }
+
+ private static RowType getKeyType(boolean hasPk) {
+ return hasPk
+ ? RowType.of(
+ new LogicalType[] {new IntType(), new VarCharType()},
+ new String[] {"k0", "k1"})
+ : RECORD_TYPE;
+ }
+
+ private static RowType getValueType(boolean hasPk) {
+ return hasPk
+ ? RECORD_TYPE
+ : RowType.of(
+ new LogicalType[] {new BigIntType(false)}, new
String[] {"_VALUE_COUNT"});
+ }
+
+ private static RowType getPartitionType(boolean partitioned) {
+ return partitioned
+ ? RowType.of(new LogicalType[] {new VarCharType()}, new
String[] {"k1"})
+ : RowType.of();
+ }
+}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/stats/FieldStats.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/stats/FieldStats.java
index 62843bf..55a0742 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/stats/FieldStats.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/stats/FieldStats.java
@@ -29,7 +29,7 @@ public class FieldStats {
@Nullable private final Object maxValue;
private final long nullCount;
- public FieldStats(Object minValue, Object maxValue, long nullCount) {
+ public FieldStats(@Nullable Object minValue, @Nullable Object maxValue,
long nullCount) {
this.minValue = minValue;
this.maxValue = maxValue;
this.nullCount = nullCount;
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/PartitionedManifestMeta.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/PartitionedManifestMeta.java
new file mode 100644
index 0000000..4968642
--- /dev/null
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/utils/PartitionedManifestMeta.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.store.file.utils;
+
+import org.apache.flink.core.memory.DataInputViewStreamWrapper;
+import org.apache.flink.core.memory.DataOutputViewStreamWrapper;
+import org.apache.flink.table.data.binary.BinaryRowData;
+import org.apache.flink.table.runtime.typeutils.BinaryRowDataSerializer;
+import org.apache.flink.table.store.file.data.DataFileMeta;
+import org.apache.flink.table.store.file.data.DataFileMetaSerializer;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.Preconditions;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/** Manifest entries per partitioned with the corresponding snapshot id. */
+public class PartitionedManifestMeta implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ /** The latest snapshot id seen at planning phase when manual compaction
is triggered. */
+ private final Long snapshotId;
+
+ /** The manifest entries collected at planning phase when manual
compaction is triggered. */
+ private transient Map<BinaryRowData, Map<Integer, List<DataFileMeta>>>
manifestEntries;
+
+ private final int partFieldCount;
+ private final RowType keyType;
+ private final RowType valueType;
+
+ public PartitionedManifestMeta(
+ Long snapshotId,
+ Map<BinaryRowData, Map<Integer, List<DataFileMeta>>>
specifiedManifestEntries,
+ int partFieldCount,
+ RowType keyType,
+ RowType valueType) {
+ Preconditions.checkNotNull(snapshotId, "Specified snapshot should not
be null.");
+ Preconditions.checkNotNull(
+ specifiedManifestEntries, "Specified manifest entries should
not be null.");
+ this.snapshotId = snapshotId;
+ this.manifestEntries = specifiedManifestEntries;
+ this.partFieldCount = partFieldCount;
+ this.keyType = keyType;
+ this.valueType = valueType;
+ }
+
+ public Long getSnapshotId() {
+ return snapshotId;
+ }
+
+ public Map<BinaryRowData, Map<Integer, List<DataFileMeta>>>
getManifestEntries() {
+ return manifestEntries;
+ }
+
+ private void writeObject(ObjectOutputStream out) throws IOException {
+ out.defaultWriteObject();
+
+ BinaryRowDataSerializer partSerializer = new
BinaryRowDataSerializer(partFieldCount);
+ DataFileMetaSerializer metaSerializer = new
DataFileMetaSerializer(keyType, valueType);
+ DataOutputViewStreamWrapper view = new
DataOutputViewStreamWrapper(out);
+ view.writeInt(manifestEntries.size());
+ for (Map.Entry<BinaryRowData, Map<Integer, List<DataFileMeta>>>
partEntry :
+ manifestEntries.entrySet()) {
+ partSerializer.serialize(partEntry.getKey(), view);
+ Map<Integer, List<DataFileMeta>> bucketEntry =
partEntry.getValue();
+ view.writeInt(bucketEntry.size());
+ for (Map.Entry<Integer, List<DataFileMeta>> entry :
bucketEntry.entrySet()) {
+ view.writeInt(entry.getKey());
+ view.writeInt(entry.getValue().size());
+ for (DataFileMeta meta : entry.getValue()) {
+ metaSerializer.serialize(meta, view);
+ }
+ }
+ }
+ }
+
+ private void readObject(ObjectInputStream in) throws
ClassNotFoundException, IOException {
+ in.defaultReadObject();
+
+ manifestEntries = new HashMap<>();
+ BinaryRowDataSerializer partSerializer = new
BinaryRowDataSerializer(partFieldCount);
+ DataFileMetaSerializer metaSerializer = new
DataFileMetaSerializer(keyType, valueType);
+ DataInputViewStreamWrapper view = new DataInputViewStreamWrapper(in);
+ int partitionNum = view.readInt();
+ for (int i = 0; i < partitionNum; i++) {
+ BinaryRowData partition = partSerializer.deserialize(view);
+ Map<Integer, List<DataFileMeta>> bucketEntry = new HashMap<>();
+ int bucketNum = view.readInt();
+ for (int j = 0; j < bucketNum; j++) {
+ int bucket = view.readInt();
+ int entryNum = view.readInt();
+ if (entryNum == 0) {
+ bucketEntry.put(bucket, Collections.emptyList());
+ } else {
+ List<DataFileMeta> metas = new ArrayList<>();
+ for (int k = 0; k < entryNum; k++) {
+ metas.add(metaSerializer.deserialize(view));
+ }
+ bucketEntry.put(bucket, metas);
+ }
+ }
+ manifestEntries.put(partition, bucketEntry);
+ }
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof PartitionedManifestMeta)) {
+ return false;
+ }
+ PartitionedManifestMeta that = (PartitionedManifestMeta) o;
+ return partFieldCount == that.partFieldCount
+ && snapshotId.equals(that.snapshotId)
+ && manifestEntries.equals(that.manifestEntries)
+ && keyType.equals(that.keyType)
+ && valueType.equals(that.valueType);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(snapshotId, manifestEntries, partFieldCount,
keyType, valueType);
+ }
+}
