Skip to content

Commit c3ef7f3

Browse files
baeminbobaeminbo
authored
Update Beam Protobuf Schema (Java) (#35150)
1 parent 39b1b26 commit c3ef7f3

30 files changed

Lines changed: 2044 additions & 919 deletions

File tree

‎buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1439,6 +1439,8 @@ class BeamModulePlugin implements Plugin<Project> {
14391439
include 'src/*/java/**/*.java'
14401440
exclude '**/DefaultPackageTest.java'
14411441
}
1442+
// For spotless:off and spotless:on
1443+
toggleOffOn()
14421444
}
14431445
}
14441446

‎sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/FieldValueHaver.java‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,33 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.beam.sdk.schemas;
19+
20+
import java.io.Serializable;
21+
import org.apache.beam.sdk.annotations.Internal;
22+
23+
/**
24+
* <b><i>For internal use only; no backwards-compatibility guarantees.</i></b>
25+
*
26+
* <p>An interface to check a field presence.
27+
*/
28+
@Internal
29+
public interface FieldValueHaver<ObjectT> extends Serializable {
30+
boolean has(ObjectT object);
31+
32+
String name();
33+
}

‎sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/GetterBasedSchemaProvider.java‎

Lines changed: 9 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,8 @@
1717
*/
1818
package org.apache.beam.sdk.schemas;
1919

20+
import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
21+
2022
import java.util.ArrayList;
2123
import java.util.Collection;
2224
import java.util.List;
@@ -405,12 +407,17 @@ Object convert(OneOfType.Value value) {
405407

406408
@NonNull
407409
FieldValueGetter<@NonNull Object, Object> converter =
408-
Verify.verifyNotNull(
410+
checkStateNotNull(
409411
converters.get(caseType.getValue()),
410412
"Missing OneOf converter for case %s.",
411413
caseType);
412414

413-
return oneOfType.createValue(caseType, converter.get(value.getValue()));
415+
Object convertedValue =
416+
checkStateNotNull(
417+
converter.get(value.getValue()),
418+
"Bug! converting a non-null value in a OneOf resulted in null result value");
419+
420+
return oneOfType.createValue(caseType, convertedValue);
414421
}
415422
}
416423

‎sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/Schema.java‎

Lines changed: 1 addition & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -492,21 +492,7 @@ private boolean equivalent(Schema other, EquivalenceNullablePolicy nullablePolic
492492

493493
@Override
494494
public String toString() {
495-
StringBuilder builder = new StringBuilder();
496-
builder.append("Fields:");
497-
builder.append(System.lineSeparator());
498-
for (Field field : fields) {
499-
builder.append(field);
500-
builder.append(System.lineSeparator());
501-
}
502-
builder.append("Encoding positions:");
503-
builder.append(System.lineSeparator());
504-
builder.append(encodingPositions);
505-
builder.append(System.lineSeparator());
506-
builder.append("Options:");
507-
builder.append(options);
508-
builder.append("UUID: " + uuid);
509-
return builder.toString();
495+
return SchemaUtils.toPrettyString(this);
510496
}
511497

512498
@Override

‎sdks/java/core/src/main/java/org/apache/beam/sdk/schemas/SchemaUtils.java‎

Lines changed: 279 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,14 +17,21 @@
1717
*/
1818
package org.apache.beam.sdk.schemas;
1919

20+
import java.util.Arrays;
21+
import java.util.List;
22+
import java.util.Map;
23+
import java.util.Objects;
2024
import org.apache.beam.sdk.schemas.Schema.FieldType;
2125
import org.apache.beam.sdk.schemas.Schema.LogicalType;
26+
import org.apache.beam.sdk.values.Row;
2227

2328
/** A set of utility functions for schemas. */
2429
@SuppressWarnings({
2530
"nullness" // TODO(https://github.com/apache/beam/issues/20497)
2631
})
2732
public class SchemaUtils {
33+
private static final String INDENT = " ";
34+
2835
/**
2936
* Given two schema that have matching types, return a nullable-widened schema.
3037
*
@@ -122,4 +129,276 @@ public static <BaseT, InputT> InputT toLogicalInputType(
122129
LogicalType<InputT, BaseT> logicalType, BaseT baseType) {
123130
return logicalType.toInputType(baseType);
124131
}
132+
133+
public static String toPrettyString(Row row) {
134+
return toPrettyRowString(row, "");
135+
}
136+
137+
public static String toPrettyString(Schema schema) {
138+
return toPrettySchemaString(schema, "");
139+
}
140+
141+
static String toFieldTypeNameString(FieldType fieldType) {
142+
return fieldType.getTypeName()
143+
+ (Boolean.TRUE.equals(fieldType.getNullable()) ? "" : " NOT NULL");
144+
}
145+
146+
static String toPrettyFieldTypeString(Schema.FieldType fieldType, String prefix) {
147+
String nextPrefix = prefix + INDENT;
148+
switch (fieldType.getTypeName()) {
149+
case BYTE:
150+
case INT16:
151+
case INT32:
152+
case INT64:
153+
case DECIMAL:
154+
case FLOAT:
155+
case DOUBLE:
156+
case STRING:
157+
case DATETIME:
158+
case BOOLEAN:
159+
case BYTES:
160+
return "<" + toFieldTypeNameString(fieldType) + ">";
161+
case ARRAY:
162+
case ITERABLE:
163+
{
164+
StringBuilder sb = new StringBuilder();
165+
sb.append("<").append(toFieldTypeNameString(fieldType)).append("> {\n");
166+
sb.append(nextPrefix)
167+
.append("<element>: ")
168+
.append(
169+
toPrettyFieldTypeString(
170+
Objects.requireNonNull(fieldType.getCollectionElementType()), nextPrefix))
171+
.append("\n");
172+
sb.append(prefix).append("}");
173+
return sb.toString();
174+
}
175+
case MAP:
176+
{
177+
StringBuilder sb = new StringBuilder();
178+
sb.append("<").append(toFieldTypeNameString(fieldType)).append("> {\n");
179+
sb.append(nextPrefix)
180+
.append("<key>: ")
181+
.append(
182+
toPrettyFieldTypeString(
183+
Objects.requireNonNull(fieldType.getMapKeyType()), nextPrefix))
184+
.append(",\n");
185+
sb.append(nextPrefix)
186+
.append("<value>: ")
187+
.append(
188+
toPrettyFieldTypeString(
189+
Objects.requireNonNull(fieldType.getMapValueType()), nextPrefix))
190+
.append("\n");
191+
sb.append(prefix).append("}");
192+
return sb.toString();
193+
}
194+
case ROW:
195+
{
196+
return "<"
197+
+ toFieldTypeNameString(fieldType)
198+
+ "> "
199+
+ toPrettySchemaString(Objects.requireNonNull(fieldType.getRowSchema()), prefix);
200+
}
201+
case LOGICAL_TYPE:
202+
{
203+
Schema.FieldType baseType =
204+
Objects.requireNonNull(fieldType.getLogicalType()).getBaseType();
205+
StringBuilder sb = new StringBuilder();
206+
sb.append("<")
207+
.append(toFieldTypeNameString(fieldType))
208+
.append("(")
209+
.append(fieldType.getLogicalType().getIdentifier())
210+
.append(")> {\n");
211+
sb.append(nextPrefix)
212+
.append("<base>: ")
213+
.append(toPrettyFieldTypeString(baseType, nextPrefix))
214+
.append("\n");
215+
sb.append(prefix).append("}");
216+
return sb.toString();
217+
}
218+
default:
219+
throw new UnsupportedOperationException(fieldType.getTypeName() + " is not supported");
220+
}
221+
}
222+
223+
static String toPrettyOptionsString(Schema.Options options, String prefix) {
224+
String nextPrefix = prefix + INDENT;
225+
StringBuilder sb = new StringBuilder();
226+
sb.append("{\n");
227+
for (String optionName : options.getOptionNames()) {
228+
sb.append(nextPrefix)
229+
.append(optionName)
230+
.append(" = ")
231+
.append(
232+
toPrettyFieldValueString(
233+
options.getType(optionName), options.getValue(optionName), nextPrefix))
234+
.append("\n");
235+
}
236+
sb.append(prefix).append("}");
237+
return sb.toString();
238+
}
239+
240+
static String toPrettyFieldValueString(Schema.FieldType fieldType, Object value, String prefix) {
241+
String nextPrefix = prefix + INDENT;
242+
switch (fieldType.getTypeName()) {
243+
case BYTE:
244+
case INT16:
245+
case INT32:
246+
case INT64:
247+
case DECIMAL:
248+
case FLOAT:
249+
case DOUBLE:
250+
case DATETIME:
251+
case BOOLEAN:
252+
return Objects.toString(value);
253+
case STRING:
254+
{
255+
String string = (String) value;
256+
return "\"" + string.replace("\\", "\\\\").replace("\"", "\\\"") + "\"";
257+
}
258+
case BYTES:
259+
{
260+
byte[] bytes = (byte[]) value;
261+
return Arrays.toString(bytes);
262+
}
263+
case ARRAY:
264+
case ITERABLE:
265+
{
266+
if (!(value instanceof List)) {
267+
throw new IllegalArgumentException(
268+
String.format(
269+
"value type is '%s' for field type '%s'",
270+
value.getClass(), fieldType.getTypeName()));
271+
}
272+
FieldType elementType = Objects.requireNonNull(fieldType.getCollectionElementType());
273+
274+
@SuppressWarnings("unchecked")
275+
List<Object> list = (List<Object>) value;
276+
if (list.isEmpty()) {
277+
return "[]";
278+
}
279+
StringBuilder sb = new StringBuilder();
280+
sb.append("[\n");
281+
int size = list.size();
282+
int index = 0;
283+
for (Object element : list) {
284+
sb.append(nextPrefix)
285+
.append(toPrettyFieldValueString(elementType, element, nextPrefix));
286+
if (index++ < size - 1) {
287+
sb.append(",\n");
288+
} else {
289+
sb.append("\n");
290+
}
291+
}
292+
sb.append(prefix).append("]");
293+
return sb.toString();
294+
}
295+
case MAP:
296+
{
297+
if (!(value instanceof Map)) {
298+
throw new IllegalArgumentException(
299+
String.format(
300+
"value type is '%s' for field type '%s'",
301+
value.getClass(), fieldType.getTypeName()));
302+
}
303+
304+
FieldType keyType = Objects.requireNonNull(fieldType.getMapKeyType());
305+
FieldType valueType = Objects.requireNonNull(fieldType.getMapValueType());
306+
307+
@SuppressWarnings("unchecked")
308+
Map<Object, Object> map = (Map<Object, Object>) value;
309+
if (map.isEmpty()) {
310+
return "{}";
311+
}
312+
313+
StringBuilder sb = new StringBuilder();
314+
sb.append("{\n");
315+
int size = map.size();
316+
int index = 0;
317+
for (Map.Entry<Object, Object> entry : map.entrySet()) {
318+
sb.append(nextPrefix)
319+
.append(toPrettyFieldValueString(keyType, entry.getKey(), nextPrefix))
320+
.append(": ")
321+
.append(toPrettyFieldValueString(valueType, entry.getValue(), nextPrefix));
322+
if (index++ < size - 1) {
323+
sb.append(",\n");
324+
} else {
325+
sb.append("\n");
326+
}
327+
}
328+
sb.append(prefix).append("}");
329+
return sb.toString();
330+
}
331+
case ROW:
332+
{
333+
return toPrettyRowString((Row) value, prefix);
334+
}
335+
case LOGICAL_TYPE:
336+
{
337+
@SuppressWarnings("unchecked")
338+
Schema.LogicalType<Object, Object> logicalType =
339+
(Schema.LogicalType<Object, Object>)
340+
Objects.requireNonNull(fieldType.getLogicalType());
341+
Schema.FieldType baseType = logicalType.getBaseType();
342+
Object baseValue = logicalType.toBaseType(value);
343+
return toPrettyFieldValueString(baseType, baseValue, prefix);
344+
}
345+
default:
346+
throw new UnsupportedOperationException(fieldType.getTypeName() + " is not supported");
347+
}
348+
}
349+
350+
static String toPrettySchemaString(Schema schema, String prefix) {
351+
String nextPrefix = prefix + INDENT;
352+
StringBuilder sb = new StringBuilder();
353+
sb.append("{\n");
354+
for (Schema.Field field : schema.getFields()) {
355+
sb.append(nextPrefix)
356+
.append(field.getName())
357+
.append(": ")
358+
.append(toPrettyFieldTypeString(field.getType(), nextPrefix));
359+
if (field.getOptions().hasOptions()) {
360+
sb.append(", fieldOptions = ")
361+
.append(toPrettyOptionsString(field.getOptions(), nextPrefix));
362+
}
363+
sb.append("\n");
364+
}
365+
sb.append(prefix).append("}");
366+
if (schema.getOptions().hasOptions()) {
367+
sb.append(", schemaOptions = ").append(toPrettyOptionsString(schema.getOptions(), prefix));
368+
}
369+
if (schema.getUUID() != null) {
370+
sb.append(", schemaUUID = ").append(schema.getUUID());
371+
}
372+
return sb.toString();
373+
}
374+
375+
static String toPrettyRowString(Row row, String prefix) {
376+
long nonNullFieldCount = row.getValues().stream().filter(Objects::nonNull).count();
377+
if (nonNullFieldCount == 0) {
378+
return "{}";
379+
}
380+
381+
String nextPrefix = prefix + INDENT;
382+
StringBuilder sb = new StringBuilder();
383+
sb.append("{\n");
384+
long nonNullFieldIndex = 0;
385+
for (Schema.Field field : row.getSchema().getFields()) {
386+
String fieldName = field.getName();
387+
Object fieldValue = row.getValue(fieldName);
388+
if (fieldValue == null) {
389+
continue;
390+
}
391+
sb.append(nextPrefix)
392+
.append(fieldName)
393+
.append(": ")
394+
.append(toPrettyFieldValueString(field.getType(), fieldValue, nextPrefix));
395+
if (nonNullFieldIndex++ < nonNullFieldCount - 1) {
396+
sb.append(",\n");
397+
} else {
398+
sb.append("\n");
399+
}
400+
}
401+
sb.append(prefix).append("}");
402+
return sb.toString();
403+
}
125404
}

0 commit comments

Comments
 (0)