Source code

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.parser.*;
030import org.apache.juneau.marshall.stream.*;
031
032/**
033 * Parses Apache Parquet binary data into bean collections.
034 *
035 * <p>
036 * Always returns a {@link List}&lt;T&gt;; each row in the file becomes one list element.
037 *
038 * <h5 class='section'>Media types:</h5>
039 * <p>
040 * Handles <c>Content-Type</c>: <bc>application/vnd.apache.parquet</bc>
041 */
042@SuppressWarnings({
043   "java:S110", // ParquetParser extends InputStreamParser, which sits on the shared Parser base-class chain; the depth comes from that format-family design.
044   "java:S1192", // Duplicated literals (argument/property names) read more clearly inline than as constants
045   "java:S9149", // Per-format static factories intentionally shadow the parent's.
046   "resource" // Closeable resources are owned by the caller's parser session; Eclipse JDT @Owning warning is by design.
047})
048public class ParquetParser extends InputStreamParser implements ParquetMetaProvider, RecordReadable, ArrayRecordReadable {
049
050   private static final int DEFAULT_MAX_LENGTH = ParquetParserSession.DEFAULT_MAX_LENGTH;
051   private static final int DEFAULT_MAX_COUNT = ParquetParserSession.DEFAULT_MAX_COUNT;
052   private static final int DEFAULT_MAX_INPUT_LENGTH = ParquetParserSession.DEFAULT_MAX_INPUT_LENGTH;
053   private static final int DEFAULT_MAX_DECOMPRESSED_BYTES = ParquetParserSession.DEFAULT_MAX_DECOMPRESSED_BYTES;
054
055   private final Map<BeanPropertyMeta,ParquetBeanPropertyMeta> parquetBeanPropertyMetas = new ConcurrentHashMap<>();
056   private final Map<ClassMeta<?>,ParquetClassMeta> parquetClassMetas = new ConcurrentHashMap<>();
057
058   /** Default parser. */
059   public static final ParquetParser DEFAULT = new ParquetParser(create());
060
061   /**
062    * Builder class.
063    */
064   public static class Builder extends InputStreamParser.Builder<Builder> {
065
066      private static final Cache<HashKey,ParquetParser> CACHE = Cache.of(HashKey.class, ParquetParser.class).build();
067
068      private String nullKeyString;
069      private int maxLength = DEFAULT_MAX_LENGTH;
070      private int maxCount = DEFAULT_MAX_COUNT;
071      private int maxInputLength = DEFAULT_MAX_INPUT_LENGTH;
072      private int maxDecompressedBytes = DEFAULT_MAX_DECOMPRESSED_BYTES;
073
074      /**
075       * Constructor, default settings.
076       */
077      protected Builder() {
078         consumes("application/vnd.apache.parquet");
079         nullKeyString = env("ParquetParser.nullKeyString", "<NULL>");
080         maxLength = env("ParquetParser.maxLength", DEFAULT_MAX_LENGTH);
081         maxCount = env("ParquetParser.maxCount", DEFAULT_MAX_COUNT);
082         maxInputLength = env("ParquetParser.maxInputLength", DEFAULT_MAX_INPUT_LENGTH);
083         maxDecompressedBytes = env("ParquetParser.maxDecompressedBytes", DEFAULT_MAX_DECOMPRESSED_BYTES);
084      }
085
086      protected Builder(Builder copyFrom) {
087         super(reqnn("copyFrom", copyFrom));
088         nullKeyString = copyFrom.nullKeyString;
089         maxLength = copyFrom.maxLength;
090         maxCount = copyFrom.maxCount;
091         maxInputLength = copyFrom.maxInputLength;
092         maxDecompressedBytes = copyFrom.maxDecompressedBytes;
093      }
094
095      protected Builder(ParquetParser copyFrom) {
096         super(reqnn("copyFrom", copyFrom));
097         nullKeyString = copyFrom.nullKeyString;
098         maxLength = copyFrom.maxLength;
099         maxCount = copyFrom.maxCount;
100         maxInputLength = copyFrom.maxInputLength;
101         maxDecompressedBytes = copyFrom.maxDecompressedBytes;
102      }
103
104      /**
105       * The string used when parsing Parquet MAP keys that represent <jk>null</jk>.
106       *
107       * <p>
108       * Must match {@link ParquetSerializer.Builder#nullKeyString(String)} for round-trip.
109       *
110       * @param value The placeholder string. Default is <js>"&lt;NULL&gt;"</js>.
111       * @return This object.
112       */
113      public Builder nullKeyString(String value) {
114         nullKeyString = value == null ? "<NULL>" : value;
115         return this;
116      }
117
118      /**
119       * The maximum allowed wire-declared length (in bytes) for a single Parquet page (compressed or
120       * uncompressed).
121       *
122       * <p>
123       * Guards against malformed or adversarial input where a small file declares a huge per-page byte
124       * size, which would otherwise trigger {@link OutOfMemoryError}.  Malformed page headers are
125       * reported as a clean parse error instead.
126       *
127       * @param value The maximum length in bytes.  Default is 256 MiB.  Values &le; 0 disable the cap.
128       * @return This object.
129       */
130      public Builder maxLength(int value) {
131         maxLength = value;
132         return this;
133      }
134
135      /**
136       * The maximum allowed wire-declared element count (file/row-group row counts, column-chunk value
137       * counts).
138       *
139       * <p>
140       * Guards against malformed or adversarial input where a small footer declares a huge row or value
141       * count, which would otherwise drive a huge {@code ArrayList} pre-allocation.  Malformed counts are
142       * reported as a clean parse error instead.
143       *
144       * @param value The maximum count.  Default is 10 million.  Values &le; 0 disable the cap.
145       * @return This object.
146       */
147      public Builder maxCount(int value) {
148         maxCount = value;
149         return this;
150      }
151
152      /**
153       * The maximum allowed size (in bytes) of the whole Parquet input buffered into memory before parsing.
154       *
155       * <p>
156       * Parquet's end-of-file footer and column-major layout require the entire input to be buffered before
157       * any record can be reconstructed.  This cap rejects an oversized input with a clean parse error
158       * instead of buffering an unbounded body into memory.
159       *
160       * @param value The maximum input size in bytes.  Default is 256 MiB.  Values &le; 0 disable the cap.
161       * @return This object.
162       */
163      public Builder maxInputLength(int value) {
164         maxInputLength = value;
165         return this;
166      }
167
168      /**
169       * The maximum allowed aggregate decompressed-byte total across every Parquet page read during a
170       * single parse.
171       *
172       * <p>
173       * Guards against a decompression bomb assembled from many pages that each individually pass the
174       * per-page {@link #maxLength(int)} cap: this ceiling charges every page's declared uncompressed
175       * size against a running per-parse total, so a chunk (or a whole file) whose pages sum past the
176       * ceiling is rejected with a clean parse error instead of buffering the excess.
177       *
178       * @param value The maximum aggregate decompressed-byte total.  Default is 64 MiB.  Values &le; 0 disable the cap.
179       * @return This object.
180       */
181      public Builder maxDecompressedBytes(int value) {
182         maxDecompressedBytes = value;
183         return this;
184      }
185
186      @Override /* InputStreamParser.Builder<?> */
187      public Builder copy() {
188         return new Builder(this);
189      }
190
191      @Override
192      public ParquetParser build() {
193         return cache(CACHE).build(ParquetParser.class);
194      }
195
196      @Override
197      public HashKey hashKey() {
198         return HashKey.of(super.hashKey(), nullKeyString, maxLength, maxCount, maxInputLength, maxDecompressedBytes);
199      }
200   }
201
202   /**
203    * Creates a new parser builder.
204    *
205    * @return A new builder.
206    */
207   public static Builder create() {
208      return new Builder();
209   }
210
211   final String nullKeyString;
212   private final int maxLength;
213   private final int maxCount;
214   private final int maxInputLength;
215   private final int maxDecompressedBytes;
216
217   /**
218    * Constructor.
219    *
220    * @param builder The builder for this parser.
221    */
222   public ParquetParser(Builder builder) {
223      super(builder);
224      nullKeyString = builder.nullKeyString;
225      maxLength = builder.maxLength;
226      maxCount = builder.maxCount;
227      maxInputLength = builder.maxInputLength;
228      maxDecompressedBytes = builder.maxDecompressedBytes;
229   }
230
231   /**
232    * Returns the maximum allowed wire-declared length (in bytes) for a single Parquet page.
233    *
234    * @return The maximum length in bytes.
235    */
236   public int getMaxLength() {
237      return maxLength;
238   }
239
240   /**
241    * Returns the maximum allowed wire-declared element count (row/value counts).
242    *
243    * @return The maximum count.
244    */
245   public int getMaxCount() {
246      return maxCount;
247   }
248
249   /**
250    * Returns the maximum allowed size (in bytes) of the whole Parquet input buffered before parsing.
251    *
252    * @return The maximum input size in bytes.
253    */
254   public int getMaxInputLength() {
255      return maxInputLength;
256   }
257
258   /**
259    * Returns the maximum allowed aggregate decompressed-byte total across every page read during a
260    * single parse.
261    *
262    * @return The maximum aggregate decompressed-byte total.
263    */
264   public int getMaxDecompressedBytes() {
265      return maxDecompressedBytes;
266   }
267
268   @Override /* Overridden from Context */
269   public Builder copy() {
270      return new Builder(this);
271   }
272
273   @Override
274   public ParquetParserSession.Builder createSession() {
275      return ParquetParserSession.create(this);
276   }
277
278   @Override
279   public ParquetBeanPropertyMeta getParquetBeanPropertyMeta(BeanPropertyMeta bpm) {
280      return bpm == null ? ParquetBeanPropertyMeta.DEFAULT
281         : parquetBeanPropertyMetas.computeIfAbsent(bpm, k -> new ParquetBeanPropertyMeta(k.getDelegateFor(), this));
282   }
283
284   @Override
285   public ParquetClassMeta getParquetClassMeta(ClassMeta<?> cm) {
286      return parquetClassMetas.computeIfAbsent(cm, k -> new ParquetClassMeta(k, this));
287   }
288
289   /**
290    * Convenience delegator for the whole-value {@link RecordReader} using <b>default session
291    * arguments</b>.  The real implementation lives on
292    * {@link ParquetParserSession#readRecords(Object)}.
293    *
294    * @param input The input.
295    * @return A new {@link RecordReader} cursor.
296    * @throws IOException If a problem occurred opening the underlying input.
297    */
298   @Override /* RecordReadable */
299   public RecordReader readRecords(Object input) throws IOException {
300      return ((RecordReadable) getSession()).readRecords(input);
301   }
302
303   /**
304    * Convenience delegator for the buffered array-element {@link RecordReader} using <b>default
305    * session arguments</b>.  The real implementation lives on
306    * {@link ParquetParserSession#readArrayRecords(Object)}.
307    *
308    * @param input The input.
309    * @return A buffered {@link RecordReader}.
310    * @throws IOException If a problem occurred reading the input.
311    */
312   @Override /* ArrayRecordReadable */
313   public RecordReader readArrayRecords(Object input) throws IOException {
314      return ((ArrayRecordReadable) getSession()).readArrayRecords(input);
315   }
316
317   /**
318    * The Parquet record cursor is buffered/{@link RecordAdapter}-backed, not O(1) streaming.
319    *
320    * @return Always <jk>false</jk>.
321    */
322   @Override /* RecordReadable */
323   public boolean isRecordStreaming() { return false; }
324
325   /**
326    * The Parquet array-record cursor is buffered/{@link RecordAdapter}-backed, not O(1) streaming.
327    *
328    * @return Always <jk>false</jk>.
329    */
330   @Override /* ArrayRecordReadable */
331   public boolean isArrayRecordStreaming() { return false; }
332}