Skip to content

Commit c352e6b

Browse files
sfc-gh-prsinghsfc-gh-prsingh
committed
Part-1 : Add serde for Req and Response
1 parent 930d66b commit c352e6b

20 files changed

+73
-1013
lines changed

‎api/src/main/java/org/apache/iceberg/exceptions/EntityNotFoundException.java‎

Lines changed: 0 additions & 34 deletions
This file was deleted.

‎core/src/main/java/org/apache/iceberg/ContentFileParser.java‎

Lines changed: 17 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -50,8 +50,6 @@ public class ContentFileParser {
5050
private static final String CONTENT_OFFSET = "content-offset";
5151
private static final String CONTENT_SIZE = "content-size-in-bytes";
5252

53-
private static ThreadLocal<Map<Integer, PartitionSpec>> extraSpecs = new ThreadLocal<>();
54-
5553
private ContentFileParser() {}
5654

5755
private static boolean hasPartitionData(StructLike partitionData) {
@@ -136,23 +134,33 @@ public static void toJson(ContentFile<?> contentFile, PartitionSpec spec, JsonGe
136134
generator.writeEndObject();
137135
}
138136

139-
public static ContentFile<?> fromJson(JsonNode jsonNode, PartitionSpec spec) {
137+
public static ContentFile<?> fromJson(JsonNode jsonNode, Map<Integer, PartitionSpec> specsById) {
140138
Preconditions.checkArgument(jsonNode != null, "Invalid JSON node for content file: null");
141139
Preconditions.checkArgument(
142140
jsonNode.isObject(), "Invalid JSON node for content file: non-object (%s)", jsonNode);
143-
// Preconditions.checkArgument(spec != null, "Invalid partition spec: null");
144-
141+
Preconditions.checkArgument(specsById != null, "Invalid partition spec: null");
145142
int specId = JsonUtil.getInt(SPEC_ID, jsonNode);
146143
FileContent fileContent = FileContent.valueOf(JsonUtil.getString(CONTENT, jsonNode));
147144
String filePath = JsonUtil.getString(FILE_PATH, jsonNode);
148145
FileFormat fileFormat = FileFormat.fromString(JsonUtil.getString(FILE_FORMAT, jsonNode));
149146

150147
PartitionData partitionData = null;
151148
if (jsonNode.has(PARTITION)) {
152-
// now its callers responsibility to set specs in the parser
153-
partitionData =
154-
partitionDataFromRawValue(
155-
jsonNode.get(PARTITION), spec == null ? extraSpecs.get().get(specId) : spec);
149+
partitionData = new PartitionData(specsById.get(specId).partitionType());
150+
StructLike structLike =
151+
(StructLike)
152+
SingleValueParser.fromJson(
153+
specsById.get(specId).partitionType(), jsonNode.get(PARTITION));
154+
Preconditions.checkState(
155+
partitionData.size() == structLike.size(),
156+
"Invalid partition data size: expected = %s, actual = %s",
157+
partitionData.size(),
158+
structLike.size());
159+
for (int pos = 0; pos < partitionData.size(); ++pos) {
160+
Class<?> javaClass =
161+
specsById.get(specId).partitionType().fields().get(pos).type().typeId().javaClass();
162+
partitionData.set(pos, structLike.get(pos, javaClass));
163+
}
156164
}
157165

158166
long fileSizeInBytes = JsonUtil.getLong(FILE_SIZE, jsonNode);
@@ -197,32 +205,6 @@ public static ContentFile<?> fromJson(JsonNode jsonNode, PartitionSpec spec) {
197205
}
198206
}
199207

200-
public static void setSpec(Map<Integer, PartitionSpec> partitionSpec) {
201-
// sets a thread local
202-
extraSpecs.set(partitionSpec);
203-
}
204-
205-
static PartitionData partitionDataFromRawValue(JsonNode rawPartitionValue, PartitionSpec spec) {
206-
if (rawPartitionValue == null) {
207-
return null;
208-
}
209-
210-
PartitionData partitionData = new PartitionData(spec.partitionType());
211-
StructLike structLike =
212-
(StructLike) SingleValueParser.fromJson(spec.partitionType(), rawPartitionValue);
213-
Preconditions.checkState(
214-
partitionData.size() == structLike.size(),
215-
"Invalid partition data size: expected = %s, actual = %s",
216-
partitionData.size(),
217-
structLike.size());
218-
for (int pos = 0; pos < partitionData.size(); ++pos) {
219-
Class<?> javaClass = spec.partitionType().fields().get(pos).type().typeId().javaClass();
220-
partitionData.set(pos, structLike.get(pos, javaClass));
221-
}
222-
223-
return partitionData;
224-
}
225-
226208
private static void metricsToJson(ContentFile<?> contentFile, JsonGenerator generator)
227209
throws IOException {
228210
generator.writeNumberField(RECORD_COUNT, contentFile.recordCount());

‎core/src/main/java/org/apache/iceberg/DataTaskParser.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import com.fasterxml.jackson.core.JsonGenerator;
2222
import com.fasterxml.jackson.databind.JsonNode;
2323
import java.io.IOException;
24+
import java.util.Map;
2425
import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
2526
import org.apache.iceberg.util.JsonUtil;
2627

@@ -64,7 +65,8 @@ static StaticDataTask fromJson(JsonNode jsonNode) {
6465
DataFile metadataFile =
6566
(DataFile)
6667
ContentFileParser.fromJson(
67-
JsonUtil.get(METADATA_FILE, jsonNode), PartitionSpec.unpartitioned());
68+
JsonUtil.get(METADATA_FILE, jsonNode),
69+
Map.of(PartitionSpec.unpartitioned().specId(), PartitionSpec.unpartitioned()));
6870

6971
JsonNode rowsArray = JsonUtil.get(ROWS, jsonNode);
7072
Preconditions.checkArgument(

‎core/src/main/java/org/apache/iceberg/FileScanTaskParser.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import com.fasterxml.jackson.core.JsonGenerator;
2222
import com.fasterxml.jackson.databind.JsonNode;
2323
import java.io.IOException;
24+
import java.util.Map;
2425
import org.apache.iceberg.expressions.Expression;
2526
import org.apache.iceberg.expressions.ExpressionParser;
2627
import org.apache.iceberg.expressions.Expressions;
@@ -86,7 +87,9 @@ static FileScanTask fromJson(JsonNode jsonNode, boolean caseSensitive) {
8687

8788
DataFile dataFile = null;
8889
if (jsonNode.has(DATA_FILE)) {
89-
dataFile = (DataFile) ContentFileParser.fromJson(jsonNode.get(DATA_FILE), spec);
90+
dataFile =
91+
(DataFile)
92+
ContentFileParser.fromJson(jsonNode.get(DATA_FILE), Map.of(spec.specId(), spec));
9093
}
9194

9295
long start = JsonUtil.getLong(START, jsonNode);
@@ -102,7 +105,8 @@ static FileScanTask fromJson(JsonNode jsonNode, boolean caseSensitive) {
102105
// parse the schema array
103106
ImmutableList.Builder<DeleteFile> builder = ImmutableList.builder();
104107
for (JsonNode deleteFileNode : deletesArray) {
105-
DeleteFile deleteFile = (DeleteFile) ContentFileParser.fromJson(deleteFileNode, spec);
108+
DeleteFile deleteFile =
109+
(DeleteFile) ContentFileParser.fromJson(deleteFileNode, Map.of(spec.specId(), spec));
106110
builder.add(deleteFile);
107111
}
108112

‎core/src/main/java/org/apache/iceberg/RESTFileScanTaskParser.java‎

Lines changed: 10 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -39,9 +39,6 @@ public class RESTFileScanTaskParser {
3939

4040
private RESTFileScanTaskParser() {}
4141

42-
private static ThreadLocal<Map<Integer, PartitionSpec>> extraSpecs = new ThreadLocal<>();
43-
private static ThreadLocal<Boolean> extraCaseSensitive = new ThreadLocal<>();
44-
4542
public static void toJson(
4643
FileScanTask fileScanTask,
4744
Set<Integer> deleteFileReferences,
@@ -65,14 +62,18 @@ public static void toJson(
6562
generator.writeEndObject();
6663
}
6764

68-
public static FileScanTask fromJson(JsonNode jsonNode, List<DeleteFile> allDeleteFiles) {
65+
public static FileScanTask fromJson(
66+
JsonNode jsonNode,
67+
List<DeleteFile> allDeleteFiles,
68+
Map<Integer, PartitionSpec> specsById,
69+
boolean isCaseSensitive) {
6970
Preconditions.checkArgument(jsonNode != null, "Invalid JSON node for file scan task: null");
7071
Preconditions.checkArgument(
7172
jsonNode.isObject(), "Invalid JSON node for file scan task: non-object (%s)", jsonNode);
7273

7374
DataFile dataFile =
74-
(DataFile) ContentFileParser.fromJson(JsonUtil.get(DATA_FILE, jsonNode), null);
75-
// get spec id of the file
75+
(DataFile) ContentFileParser.fromJson(JsonUtil.get(DATA_FILE, jsonNode), specsById);
76+
// specId from the dataFile
7677
int specId = dataFile.specId();
7778

7879
DeleteFile[] deleteFiles = null;
@@ -90,16 +91,11 @@ public static FileScanTask fromJson(JsonNode jsonNode, List<DeleteFile> allDelet
9091
filter = ExpressionParser.fromJson(jsonNode.get(RESIDUAL));
9192
}
9293

93-
String schemaString = SchemaParser.toJson(extraSpecs.get().get(specId).schema());
94-
String specString = PartitionSpecParser.toJson(extraSpecs.get().get(specId));
94+
String schemaString = SchemaParser.toJson(specsById.get(specId).schema());
95+
String specString = PartitionSpecParser.toJson(specsById.get(specId));
9596
ResidualEvaluator boundResidual =
96-
ResidualEvaluator.of(extraSpecs.get().get(specId), filter, extraCaseSensitive.get());
97+
ResidualEvaluator.of(specsById.get(specId), filter, isCaseSensitive);
9798

9899
return new BaseFileScanTask(dataFile, deleteFiles, schemaString, specString, boundResidual);
99100
}
100-
101-
public static void setExtraInfo(Map<Integer, PartitionSpec> spec, boolean caseSensitive) {
102-
extraSpecs.set(spec);
103-
extraCaseSensitive.set(caseSensitive);
104-
}
105101
}

‎core/src/main/java/org/apache/iceberg/RESTPlanningMode.java‎

Lines changed: 0 additions & 47 deletions
This file was deleted.

‎core/src/main/java/org/apache/iceberg/RESTTable.java‎

Lines changed: 0 additions & 71 deletions
This file was deleted.

0 commit comments

Comments
 (0)