Skip to content

Commit 7676575

Browse files
durgaprasadmldurgaprasadml
committed
Spotless apply for Delta IO Integration Tests
1 parent 47a7fb4 commit 7676575

1 file changed

Lines changed: 101 additions & 70 deletions

File tree

  • sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta

‎sdks/java/io/delta/src/test/java/org/apache/beam/sdk/io/delta/DeltaIOIT.java‎

Lines changed: 101 additions & 70 deletions
Original file line numberDiff line numberDiff line change
@@ -24,10 +24,8 @@
2424
import java.nio.channels.Channels;
2525
import java.util.ArrayList;
2626
import java.util.Collections;
27-
import java.util.HashMap;
2827
import java.util.HashSet;
2928
import java.util.List;
30-
import java.util.Map;
3129
import java.util.Set;
3230
import java.util.UUID;
3331
import java.util.function.Function;
@@ -148,7 +146,8 @@ public void testReadPartitioned() throws Exception {
148146
runIntegrationTest(1000, true, "partitioned");
149147
}
150148

151-
private void runIntegrationTest(int numRecords, boolean isPartitioned, String scenarioName) throws Exception {
149+
private void runIntegrationTest(int numRecords, boolean isPartitioned, String scenarioName)
150+
throws Exception {
152151
String tablePath = appendTimestampSuffix(tablePathPrefix + "-" + scenarioName);
153152
try {
154153
// 1. Write Parquet files using Beam
@@ -174,59 +173,78 @@ private void writeParquetFiles(String tablePath, int numRecords, boolean isParti
174173
pipelineWrite
175174
.apply("Generate sequence", GenerateSequence.from(0).to(numRecords))
176175
.apply("Construct TestRows", ParDo.of(new TestRow.DeterministicallyConstructTestRowFn()))
177-
.apply("Convert to Partitioned Avro", ParDo.of(new DoFn<TestRow, GenericRecord>() {
178-
@ProcessElement
179-
public void processElement(ProcessContext c) {
180-
TestRow row = c.element();
181-
String part = (row.id() % 2 == 0) ? "even" : "odd";
182-
c.output(new GenericRecordBuilder(PARTITIONED_AVRO_SCHEMA)
183-
.set("id", row.id())
184-
.set("name", row.name())
185-
.set("part", part)
186-
.build());
187-
}
188-
}))
176+
.apply(
177+
"Convert to Partitioned Avro",
178+
ParDo.of(
179+
new DoFn<TestRow, GenericRecord>() {
180+
@ProcessElement
181+
public void processElement(ProcessContext c) {
182+
TestRow row = c.element();
183+
String part = (row.id() % 2 == 0) ? "even" : "odd";
184+
c.output(
185+
new GenericRecordBuilder(PARTITIONED_AVRO_SCHEMA)
186+
.set("id", row.id())
187+
.set("name", row.name())
188+
.set("part", part)
189+
.build());
190+
}
191+
}))
189192
.setCoder(AvroCoder.of(PARTITIONED_AVRO_SCHEMA))
190-
.apply("Write Partitioned Parquet", FileIO.<String, GenericRecord>writeDynamic()
191-
.by(record -> "part=" + record.get("part"))
192-
.via(ParquetIO.sink(PARTITIONED_AVRO_SCHEMA))
193-
.to(tablePath)
194-
.withNaming(key -> DefaultFilenamePolicy.fromStandardNaming(
195-
key + "/part", null, ".parquet", false)));
193+
.apply(
194+
"Write Partitioned Parquet",
195+
FileIO.<String, GenericRecord>writeDynamic()
196+
.by(record -> "part=" + record.get("part"))
197+
.via(ParquetIO.sink(PARTITIONED_AVRO_SCHEMA))
198+
.to(tablePath)
199+
.withNaming(
200+
key ->
201+
DefaultFilenamePolicy.fromStandardNaming(
202+
key + "/part", null, ".parquet", false)));
196203
} else {
197204
pipelineWrite
198205
.apply("Generate sequence", GenerateSequence.from(0).to(numRecords))
199206
.apply("Construct TestRows", ParDo.of(new TestRow.DeterministicallyConstructTestRowFn()))
200-
.apply("Convert to Avro", ParDo.of(new DoFn<TestRow, GenericRecord>() {
201-
@ProcessElement
202-
public void processElement(ProcessContext c) {
203-
TestRow row = c.element();
204-
c.output(new GenericRecordBuilder(AVRO_SCHEMA)
205-
.set("id", row.id())
206-
.set("name", row.name())
207-
.build());
208-
}
209-
}))
207+
.apply(
208+
"Convert to Avro",
209+
ParDo.of(
210+
new DoFn<TestRow, GenericRecord>() {
211+
@ProcessElement
212+
public void processElement(ProcessContext c) {
213+
TestRow row = c.element();
214+
c.output(
215+
new GenericRecordBuilder(AVRO_SCHEMA)
216+
.set("id", row.id())
217+
.set("name", row.name())
218+
.build());
219+
}
220+
}))
210221
.setCoder(AvroCoder.of(AVRO_SCHEMA))
211-
.apply("Write Parquet files", FileIO.<GenericRecord>write()
212-
.via(ParquetIO.sink(AVRO_SCHEMA))
213-
.to(tablePath)
214-
.withNumShards(2));
222+
.apply(
223+
"Write Parquet files",
224+
FileIO.<GenericRecord>write()
225+
.via(ParquetIO.sink(AVRO_SCHEMA))
226+
.to(tablePath)
227+
.withNumShards(2));
215228
}
216229
pipelineWrite.run().waitUntilFinish();
217230
}
218231

219232
private void generateDeltaLog(String tablePath, boolean isPartitioned) throws Exception {
220-
List<MatchResult.Metadata> metadataList = FileSystems.match(tablePath + "/**/*.parquet").metadata();
221-
ResourceId logFileResourceId = FileSystems.matchNewResource(tablePath + "/_delta_log/00000000000000000000.json", false);
222-
223-
try (PrintWriter writer = new PrintWriter(
224-
Channels.newWriter(FileSystems.create(logFileResourceId, MimeTypes.TEXT), "UTF-8"))) {
233+
List<MatchResult.Metadata> metadataList =
234+
FileSystems.match(tablePath + "/**/*.parquet").metadata();
235+
ResourceId logFileResourceId =
236+
FileSystems.matchNewResource(tablePath + "/_delta_log/00000000000000000000.json", false);
237+
238+
try (PrintWriter writer =
239+
new PrintWriter(
240+
Channels.newWriter(FileSystems.create(logFileResourceId, MimeTypes.TEXT), "UTF-8"))) {
225241
writer.println("{\"protocol\":{\"minReaderVersion\":1,\"minWriterVersion\":1}}");
226242
if (isPartitioned) {
227-
writer.println("{\"metaData\":{\"id\":\"test-uuid-partitioned\",\"format\":{\"provider\":\"parquet\",\"options\":{}},\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"part\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}]}\",\"partitionBy\":[\"part\"],\"configuration\":{},\"createdTime\":1717081200000}}");
243+
writer.println(
244+
"{\"metaData\":{\"id\":\"test-uuid-partitioned\",\"format\":{\"provider\":\"parquet\",\"options\":{}},\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"part\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}]}\",\"partitionBy\":[\"part\"],\"configuration\":{},\"createdTime\":1717081200000}}");
228245
} else {
229-
writer.println("{\"metaData\":{\"id\":\"test-uuid-nonpartitioned\",\"format\":{\"provider\":\"parquet\",\"options\":{}},\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}]}\",\"partitionBy\":[],\"configuration\":{},\"createdTime\":1717081200000}}");
246+
writer.println(
247+
"{\"metaData\":{\"id\":\"test-uuid-nonpartitioned\",\"format\":{\"provider\":\"parquet\",\"options\":{}},\"schemaString\":\"{\\\"type\\\":\\\"struct\\\",\\\"fields\\\":[{\\\"name\\\":\\\"id\\\",\\\"type\\\":\\\"integer\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}},{\\\"name\\\":\\\"name\\\",\\\"type\\\":\\\"string\\\",\\\"nullable\\\":true,\\\"metadata\\\":{}}]}\",\"partitionBy\":[],\"configuration\":{},\"createdTime\":1717081200000}}");
230248
}
231249

232250
for (MatchResult.Metadata metadata : metadataList) {
@@ -241,48 +259,60 @@ private void generateDeltaLog(String tablePath, boolean isPartitioned) throws Ex
241259
String addAction;
242260
if (isPartitioned) {
243261
String partValue = relativePath.contains("part=even") ? "even" : "odd";
244-
addAction = String.format(
245-
"{\"add\":{\"path\":\"%s\",\"partitionValues\":{\"part\":\"%s\"},\"size\":%d,\"modificationTime\":%d,\"dataChange\":true}}",
246-
relativePath, partValue, metadata.sizeBytes(), metadata.lastModifiedMillis());
262+
addAction =
263+
String.format(
264+
"{\"add\":{\"path\":\"%s\",\"partitionValues\":{\"part\":\"%s\"},\"size\":%d,\"modificationTime\":%d,\"dataChange\":true}}",
265+
relativePath, partValue, metadata.sizeBytes(), metadata.lastModifiedMillis());
247266
} else {
248-
addAction = String.format(
249-
"{\"add\":{\"path\":\"%s\",\"partitionValues\":{},\"size\":%d,\"modificationTime\":%d,\"dataChange\":true}}",
250-
relativePath, metadata.sizeBytes(), metadata.lastModifiedMillis());
267+
addAction =
268+
String.format(
269+
"{\"add\":{\"path\":\"%s\",\"partitionValues\":{},\"size\":%d,\"modificationTime\":%d,\"dataChange\":true}}",
270+
relativePath, metadata.sizeBytes(), metadata.lastModifiedMillis());
251271
}
252272
writer.println(addAction);
253273
}
254274
}
255275
}
256276

257-
private void readAndVerify(String tablePath, int numRecords, boolean isPartitioned, String scenarioName) {
277+
private void readAndVerify(
278+
String tablePath, int numRecords, boolean isPartitioned, String scenarioName) {
258279
PCollection<org.apache.beam.sdk.values.Row> deltaRows =
259280
pipelineRead.apply("Read from Delta", DeltaIO.readRows().from(tablePath));
260281

261282
PCollection<org.apache.beam.sdk.values.Row> monitoredRows =
262-
deltaRows.apply("TimeMonitor", ParDo.of(new TimeMonitor<>(NAMESPACE, scenarioName + "_read")));
283+
deltaRows.apply(
284+
"TimeMonitor", ParDo.of(new TimeMonitor<>(NAMESPACE, scenarioName + "_read")));
263285

264286
PCollection<TestRow> namesAndIds;
265287
if (isPartitioned) {
266-
namesAndIds = monitoredRows.apply("Convert to TestRow", ParDo.of(new DoFn<org.apache.beam.sdk.values.Row, TestRow>() {
267-
@ProcessElement
268-
public void processElement(ProcessContext c) {
269-
org.apache.beam.sdk.values.Row r = c.element();
270-
int id = r.getInt32("id");
271-
String name = r.getString("name");
272-
String part = r.getString("part");
273-
String expectedPart = (id % 2 == 0) ? "even" : "odd";
274-
org.junit.Assert.assertEquals(expectedPart, part);
275-
c.output(TestRow.create(id, name));
276-
}
277-
}));
288+
namesAndIds =
289+
monitoredRows.apply(
290+
"Convert to TestRow",
291+
ParDo.of(
292+
new DoFn<org.apache.beam.sdk.values.Row, TestRow>() {
293+
@ProcessElement
294+
public void processElement(ProcessContext c) {
295+
org.apache.beam.sdk.values.Row r = c.element();
296+
int id = r.getInt32("id");
297+
String name = r.getString("name");
298+
String part = r.getString("part");
299+
String expectedPart = (id % 2 == 0) ? "even" : "odd";
300+
org.junit.Assert.assertEquals(expectedPart, part);
301+
c.output(TestRow.create(id, name));
302+
}
303+
}));
278304
} else {
279-
namesAndIds = monitoredRows.apply("Convert to TestRow", ParDo.of(new DoFn<org.apache.beam.sdk.values.Row, TestRow>() {
280-
@ProcessElement
281-
public void processElement(ProcessContext c) {
282-
org.apache.beam.sdk.values.Row r = c.element();
283-
c.output(TestRow.create(r.getInt32("id"), r.getString("name")));
284-
}
285-
}));
305+
namesAndIds =
306+
monitoredRows.apply(
307+
"Convert to TestRow",
308+
ParDo.of(
309+
new DoFn<org.apache.beam.sdk.values.Row, TestRow>() {
310+
@ProcessElement
311+
public void processElement(ProcessContext c) {
312+
org.apache.beam.sdk.values.Row r = c.element();
313+
c.output(TestRow.create(r.getInt32("id"), r.getString("name")));
314+
}
315+
}));
286316
}
287317

288318
PCollection<String> consolidatedHashcode =
@@ -321,7 +351,8 @@ private void collectAndPublishMetrics(PipelineResult readResult, String metricNa
321351
long end = reader.getEndTimeMetric(metricName);
322352
double duration = (end - start) / 1e3;
323353
double throughput = duration > 0 ? records / duration : 0.0;
324-
return NamedTestResult.create(uuid, timestamp, metricName + "_throughput_ops_sec", throughput);
354+
return NamedTestResult.create(
355+
uuid, timestamp, metricName + "_throughput_ops_sec", throughput);
325356
});
326357

327358
IOITMetrics metrics = new IOITMetrics(suppliers, readResult, NAMESPACE, uuid, timestamp);

0 commit comments

Comments
 (0)