001/* 002 * Licensed to the Apache Software Foundation (ASF) under one or more 003 * contributor license agreements. See the NOTICE file distributed with 004 * this work for additional information regarding copyright ownership. 005 * The ASF licenses this file to You under the Apache License, Version 2.0 006 * (the "License"); you may not use this file except in compliance with 007 * the License. You may obtain a copy of the License at 008 * 009 * http://www.apache.org/licenses/LICENSE-2.0 010 * 011 * Unless required by applicable law or agreed to in writing, software 012 * distributed under the License is distributed on an "AS IS" BASIS, 013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 014 * See the License for the specific language governing permissions and 015 * limitations under the License. 016 */ 017package org.apache.juneau.marshall.parquet; 018 019import static org.apache.juneau.commons.utils.Shorts.*; 020import static org.apache.juneau.commons.utils.SystemUtils.*; 021 022import java.io.*; 023import java.util.*; 024import java.util.concurrent.*; 025 026import org.apache.juneau.commons.bean.*; 027import org.apache.juneau.commons.collections.*; 028import org.apache.juneau.marshall.*; 029import org.apache.juneau.marshall.serializer.*; 030import org.apache.juneau.marshall.stream.*; 031 032/** 033 * Serializes bean collections to Apache Parquet binary format. 034 * 035 * <p> 036 * Parquet is a columnar format; each bean is one row and each bean property is one column. 037 * The serializer accepts a {@link Collection}, array, or single bean at the root. 038 * 039 * <h5 class='section'>Media types:</h5> 040 * <p> 041 * Produces <c>Content-Type</c>: <bc>application/vnd.apache.parquet</bc> 042 */ 043@SuppressWarnings({ 044 "java:S110", // Inheritance depth acceptable for this class hierarchy 045 "java:S1192", // Duplicated literals (argument/property names) read more clearly inline than as constants 046 "java:S9149", // Per-format static factories intentionally shadow the parent's. 047 "resource" // Closeable resources are owned by the caller's serializer session; Eclipse JDT @Owning warning is by design. 048}) 049public class ParquetSerializer extends OutputStreamSerializer implements ParquetMetaProvider, RecordWritable, ArrayRecordWritable { 050 051 /** Default serializer. */ 052 public static final ParquetSerializer DEFAULT = new ParquetSerializer(create()); 053 054 /** 055 * Builder class. 056 */ 057 public static class Builder extends OutputStreamSerializer.Builder<Builder> { 058 059 private static final Cache<HashKey,ParquetSerializer> CACHE = Cache.of(HashKey.class, ParquetSerializer.class).build(); 060 061 private CompressionCodec compressionCodec = CompressionCodec.UNCOMPRESSED; 062 private int rowGroupSize = 128 * 1024 * 1024; 063 private int pageSize = 1024 * 1024; 064 private boolean addBeanTypesParquet = false; 065 private boolean writeDatesAsTimestamp = true; 066 private boolean emitLogicalTypes = false; 067 private boolean nativeLogicalTypes = false; 068 private ParquetCycleHandling cycleHandling = ParquetCycleHandling.NULL; 069 private int maxRecursionDepth = 5; 070 private String nullKeyString; 071 072 /** 073 * Constructor, default settings. 074 */ 075 protected Builder() { 076 produces("application/vnd.apache.parquet"); 077 accept("application/vnd.apache.parquet"); 078 emitLogicalTypes = env("ParquetSerializer.emitLogicalTypes", false); 079 maxRecursionDepth = env("ParquetSerializer.maxRecursionDepth", 5); 080 nullKeyString = env("ParquetSerializer.nullKeyString", "<NULL>"); 081 } 082 083 /** 084 * Copy constructor. 085 * 086 * @param copyFrom The builder to copy from. 087 * <br>Cannot be <jk>null</jk>. 088 */ 089 protected Builder(Builder copyFrom) { 090 super(reqnn("copyFrom", copyFrom)); 091 compressionCodec = copyFrom.compressionCodec; 092 rowGroupSize = copyFrom.rowGroupSize; 093 pageSize = copyFrom.pageSize; 094 addBeanTypesParquet = copyFrom.addBeanTypesParquet; 095 writeDatesAsTimestamp = copyFrom.writeDatesAsTimestamp; 096 emitLogicalTypes = copyFrom.emitLogicalTypes; 097 nativeLogicalTypes = copyFrom.nativeLogicalTypes; 098 cycleHandling = copyFrom.cycleHandling; 099 maxRecursionDepth = copyFrom.maxRecursionDepth; 100 nullKeyString = copyFrom.nullKeyString; 101 } 102 103 /** 104 * Copy constructor. 105 * 106 * @param copyFrom The serializer to copy from. 107 * <br>Cannot be <jk>null</jk>. 108 */ 109 protected Builder(ParquetSerializer copyFrom) { 110 super(reqnn("copyFrom", copyFrom)); 111 compressionCodec = copyFrom.compressionCodec; 112 rowGroupSize = copyFrom.rowGroupSize; 113 pageSize = copyFrom.pageSize; 114 addBeanTypesParquet = copyFrom.addBeanTypesParquet; 115 writeDatesAsTimestamp = copyFrom.writeDatesAsTimestamp; 116 emitLogicalTypes = copyFrom.emitLogicalTypes; 117 nativeLogicalTypes = copyFrom.nativeLogicalTypes; 118 cycleHandling = copyFrom.cycleHandling; 119 maxRecursionDepth = copyFrom.maxRecursionDepth; 120 nullKeyString = copyFrom.nullKeyString; 121 } 122 123 /** 124 * Sets the compression codec. 125 * 126 * @param value UNCOMPRESSED or GZIP. <jk>null</jk> resets to UNCOMPRESSED. 127 * @return This object. 128 * @throws IllegalArgumentException If the codec cannot be written (e.g. SNAPPY, which is decode-only). 129 */ 130 public Builder compressionCodec(CompressionCodec value) { 131 if (value != null && ! value.writable()) 132 throw CompressionCodec.forWriteError(value.name()); 133 compressionCodec = or(value, CompressionCodec.UNCOMPRESSED); 134 return this; 135 } 136 137 /** 138 * Sets the target row group size in bytes. 139 * 140 * @param value Size in bytes. 141 * @return This object. 142 */ 143 public Builder rowGroupSize(int value) { 144 rowGroupSize = Math.max(1024, value); 145 return this; 146 } 147 148 /** 149 * Sets the target page size in bytes. 150 * 151 * @param value Size in bytes. 152 * @return This object. 153 */ 154 public Builder pageSize(int value) { 155 pageSize = Math.max(1024, value); 156 return this; 157 } 158 159 /** 160 * Adds <js>_type</js> column for polymorphic beans. 161 * 162 * @param value Whether to add the column. 163 * @return This object. 164 */ 165 public Builder addBeanTypesParquet(boolean value) { 166 addBeanTypesParquet = value; 167 return this; 168 } 169 170 /** 171 * Sets whether dates are written as TIMESTAMP(MILLIS). 172 * 173 * @param value If true, use INT64+TIMESTAMP_MILLIS; else use ISO 8601 string. 174 * @return This object. 175 */ 176 public Builder writeDatesAsTimestamp(boolean value) { 177 writeDatesAsTimestamp = value; 178 return this; 179 } 180 181 /** 182 * Emits explicit logical-type discriminator metadata for ambiguous physical representations. 183 * 184 * <p> 185 * Disabled by default, which preserves the current Parquet wire shape (physical type plus 186 * {@code convertedType} only). Modeled on JSON's optional <js>_type</js> discriminator: when enabled, 187 * the serializer additionally writes the Parquet <c>LogicalType</c> union into the file footer for 188 * columns whose physical encoding is otherwise ambiguous. 189 * 190 * <p> 191 * First-pass scope is UUID only — the one type whose round-trip relies on the 192 * {@code FIXED_LEN_BYTE_ARRAY} physical-type signal. The legacy {@code convertedType} field is still 193 * emitted alongside the union, so readers that consume only the legacy field (including older Juneau 194 * versions) continue to work unchanged, while modern readers can route on {@code logicalType}. 195 * 196 * @param value Whether to emit logical-type discriminator metadata. 197 * @return This object. 198 */ 199 public Builder emitLogicalTypes(boolean value) { 200 emitLogicalTypes = value; 201 return this; 202 } 203 204 /** 205 * Emits binary-native logical types (DECIMAL, DATE, TIME, TIMESTAMP-micros) on write instead of the 206 * default string / TIMESTAMP-millis normalization. 207 * 208 * <p> 209 * Disabled by default, which preserves the current wire shape (and lossless Juneau↔Juneau 210 * round-trips). When enabled, {@code BigDecimal} columns are written as DECIMAL with scale/precision, 211 * {@code LocalDate} as DATE, {@code LocalTime}/{@code OffsetTime} as TIME(MICROS), and date/instant 212 * temporals as TIMESTAMP(MICROS) — interoperable with parquet-mr / Spark and sub-millisecond precise. 213 * Read-side decode of these types is always on regardless of this flag. 214 * 215 * @param value Whether to emit binary-native logical types on write. 216 * @return This object. 217 */ 218 public Builder nativeLogicalTypes(boolean value) { 219 nativeLogicalTypes = value; 220 return this; 221 } 222 223 /** 224 * Sets how to handle cyclic references during serialization. 225 * 226 * @param value THROW to fail with {@link SerializeException}; 227 * NULL to serialize cycles as <jk>null</jk>. 228 * @return This object. 229 */ 230 public Builder cycleHandling(ParquetCycleHandling value) { 231 cycleHandling = or(value, ParquetCycleHandling.NULL); 232 return this; 233 } 234 235 /** 236 * Sets the maximum expansion depth for self-referential (recursive) bean types. 237 * 238 * <p> 239 * A self-referential type (e.g. <c>class Node { String val; Node next; }</c>) is expanded into this 240 * many nested column levels so acyclic data (trees, linked lists) up to this depth round-trips 241 * losslessly instead of degrading to a {@code toString} representation (GAP-13). Structure deeper 242 * than this is truncated to <jk>null</jk> (documented-lossy). Larger values support deeper structures 243 * at the cost of a wider schema; the schema is expanded structurally regardless of the actual data 244 * depth, so keep this modest for recursive types with high fan-out. 245 * 246 * @param value The maximum recursion expansion depth. Values below 1 are clamped to 1. 247 * @return This object. 248 */ 249 public Builder maxRecursionDepth(int value) { 250 maxRecursionDepth = Math.max(1, value); 251 return this; 252 } 253 254 /** 255 * The string used to represent <jk>null</jk> map keys in Parquet MAP key_value format. 256 * 257 * <p> 258 Parquet MAP keys are UTF8; <jk>null</jk> keys cannot be stored directly. When serializing a 259 * {@link Map} with a <jk>null</jk> key, this string is used. When parsing, a key 260 * equal to this string is restored as <jk>null</jk>. 261 * 262 * @param value The placeholder string for null keys. Default is <js>"<NULL>"</js>. 263 * @return This object. 264 */ 265 public Builder nullKeyString(String value) { 266 nullKeyString = value == null ? "<NULL>" : value; 267 return this; 268 } 269 270 @Override /* OutputStreamSerializer.Builder<?> */ 271 public Builder copy() { 272 return new Builder(this); 273 } 274 275 @Override 276 public ParquetSerializer build() { 277 return cache(CACHE).build(ParquetSerializer.class); 278 } 279 280 @Override 281 public HashKey hashKey() { 282 return HashKey.of(super.hashKey(), compressionCodec, rowGroupSize, pageSize, addBeanTypesParquet, writeDatesAsTimestamp, emitLogicalTypes, nativeLogicalTypes, cycleHandling, maxRecursionDepth, nullKeyString); 283 } 284 } 285 286 /** 287 * Creates a new builder. 288 * 289 * @return A new builder. 290 */ 291 public static Builder create() { 292 return new Builder(); 293 } 294 295 final CompressionCodec compressionCodec; 296 final int rowGroupSize; 297 final int pageSize; 298 final boolean addBeanTypesParquet; 299 final boolean writeDatesAsTimestamp; 300 final boolean emitLogicalTypes; 301 final boolean nativeLogicalTypes; 302 final ParquetCycleHandling cycleHandling; 303 final int maxRecursionDepth; 304 final String nullKeyString; 305 306 private final Map<BeanPropertyMeta,ParquetBeanPropertyMeta> parquetBeanPropertyMetas = new ConcurrentHashMap<>(); 307 private final Map<ClassMeta<?>,ParquetClassMeta> parquetClassMetas = new ConcurrentHashMap<>(); 308 309 /** 310 * Constructor. 311 * 312 * @param builder The builder. 313 */ 314 public ParquetSerializer(Builder builder) { 315 super(builder); 316 compressionCodec = builder.compressionCodec; 317 rowGroupSize = builder.rowGroupSize; 318 pageSize = builder.pageSize; 319 addBeanTypesParquet = builder.addBeanTypesParquet; 320 writeDatesAsTimestamp = builder.writeDatesAsTimestamp; 321 emitLogicalTypes = builder.emitLogicalTypes; 322 nativeLogicalTypes = builder.nativeLogicalTypes; 323 cycleHandling = builder.cycleHandling; 324 maxRecursionDepth = builder.maxRecursionDepth; 325 nullKeyString = builder.nullKeyString; 326 } 327 328 @Override /* Overridden from Context */ 329 public Builder copy() { 330 return new Builder(this); 331 } 332 333 @Override 334 public ParquetSerializerSession.Builder createSession() { 335 return ParquetSerializerSession.create(this); 336 } 337 338 @Override 339 public ParquetBeanPropertyMeta getParquetBeanPropertyMeta(BeanPropertyMeta bpm) { 340 return bpm == null ? ParquetBeanPropertyMeta.DEFAULT 341 : parquetBeanPropertyMetas.computeIfAbsent(bpm, k -> new ParquetBeanPropertyMeta(k.getDelegateFor(), this)); 342 } 343 344 @Override 345 public ParquetClassMeta getParquetClassMeta(ClassMeta<?> cm) { 346 return parquetClassMetas.computeIfAbsent(cm, k -> new ParquetClassMeta(k, this)); 347 } 348 349 /** 350 * Convenience delegator for the whole-value {@link RecordWriter} using <b>default session 351 * arguments</b>. The real implementation lives on 352 * {@link ParquetSerializerSession#writeRecords(Object)}. 353 * 354 * @param output The output. 355 * @return A new {@link RecordWriter}. 356 * @throws IOException If a problem occurred opening the underlying output. 357 */ 358 @Override /* RecordWritable */ 359 public RecordWriter writeRecords(Object output) throws IOException { 360 return ((RecordWritable) getSession()).writeRecords(output); 361 } 362 363 /** 364 * Convenience delegator for the buffered array-element {@link RecordWriter} using <b>default 365 * session arguments</b>. The real implementation lives on 366 * {@link ParquetSerializerSession#writeArrayRecords(Object)}. 367 * 368 * @param output The output. 369 * @return A new {@link RecordWriter}. 370 * @throws IOException If a problem occurred opening the underlying output. 371 */ 372 @Override /* ArrayRecordWritable */ 373 public RecordWriter writeArrayRecords(Object output) throws IOException { 374 return ((ArrayRecordWritable) getSession()).writeArrayRecords(output); 375 } 376 377 /** 378 * The Parquet record writer is buffered/{@link RecordAdapter}-backed, not O(1) streaming. 379 * 380 * @return Always <jk>false</jk>. 381 */ 382 @Override /* RecordWritable */ 383 public boolean isRecordStreaming() { return false; } 384 385 /** 386 * The Parquet array-record writer is buffered/{@link RecordAdapter}-backed, not O(1) streaming. 387 * 388 * @return Always <jk>false</jk>. 389 */ 390 @Override /* ArrayRecordWritable */ 391 public boolean isArrayRecordStreaming() { return false; } 392}