27420372206

Committed 12 Jun 2026 01:58PM UTC coverage: 99.246% (-0.8%) from 100.0%

Build # 27420372206

Build Type

Pull #67

github

Committed by

web-flow

Commit Message

Merge 29689dc0b into 4b0fcbcb4

Pull Request Pull Request #67: 4.1.4: auto-detect fixes, new input formats, raw SQL suppor

Coverage Stats

70 of 94 new or added lines in 7 files covered. (74.47%)

3159 of 3183 relevant lines covered (99.25%)

22.36 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

42.86

/src/Concerns/HasInputFormat.php

<?php
declare(strict_types=1);

namespace Level23\Druid\Concerns;

use Level23\Druid\InputFormats\FlattenSpec;
use Level23\Druid\InputFormats\CsvInputFormat;
use Level23\Druid\InputFormats\TsvInputFormat;
use Level23\Druid\InputFormats\OrcInputFormat;
use Level23\Druid\InputFormats\JsonInputFormat;
use Level23\Druid\InputFormats\KafkaInputFormat;
use Level23\Druid\InputFormats\AvroOcfInputFormat;
use Level23\Druid\InputFormats\ParquetInputFormat;
use Level23\Druid\InputFormats\ProtobufInputFormat;
use Level23\Druid\InputFormats\AvroStreamInputFormat;
use Level23\Druid\InputFormats\InputFormatInterface;

trait HasInputFormat
{
    /**
     * @var \Level23\Druid\InputFormats\InputFormatInterface|null
     */
    protected ?InputFormatInterface $inputFormat = null;

    /**
     * Specify that we use JSON as input format.
     *
     * @param FlattenSpec|null        $flattenSpec Specifies flattening configuration for nested JSON data. See
     *                                             flattenSpec for more info.
     * @param array<string,bool>|null $features    JSON parser features supported by Jackson library. Those features
     *                                             will be applied when parsing the input JSON data.
     *
     * @see https://github.com/FasterXML/jackson-core/wiki/JsonParser-Features
     */
    public function jsonFormat(?FlattenSpec $flattenSpec = null, ?array $features = null): self
    {
        $this->inputFormat = new JsonInputFormat($flattenSpec, $features);

        return $this;
    }

    /**
     * Specify that we use CSV as input format.
     *
     * @param string[]|null $columns               Specifies the columns of the data. The columns should be in the same
     *                                             order with the columns of your data.
     * @param string|null   $listDelimiter         A custom delimiter for multi-value dimensions.
     * @param bool|null     $findColumnsFromHeader If this is set, the task will find the column names from the header
     *                                             row. Note that skipHeaderRows will be applied before finding column
     *                                             names from the header. For example, if you set skipHeaderRows to 2
     *                                             and findColumnsFromHeader to true, the task will skip the first two
     *                                             lines and then extract column information from the third line.
     *                                             columns will be ignored if this is set to true.
     * @param int           $skipHeaderRows        If this is set, the task will skip the first skipHeaderRows rows.
     */
    public function csvFormat(
        ?array $columns = null,
        ?string $listDelimiter = null,
        ?bool $findColumnsFromHeader = null,
        int $skipHeaderRows = 0
    ): self {
        $this->inputFormat = new CsvInputFormat($columns, $listDelimiter, $findColumnsFromHeader, $skipHeaderRows);

        return $this;
    }

    /**
     * Specify that we use TSV as input format.
     *
     * @param array<string>|null $columns               Specifies the columns of the data. The columns should be in the
     *                                                  same order with the columns of your data.
     * @param string|null        $delimiter             A custom delimiter for data values.
     * @param string|null        $listDelimiter         A custom delimiter for multi-value dimensions.
     * @param bool|null          $findColumnsFromHeader If this is set, the task will find the column names from the
     *                                                  header row. Note that skipHeaderRows will be applied before
     *                                                  finding column names from the header. For example, if you set
     *                                                  skipHeaderRows to 2 and findColumnsFromHeader to true, the task
     *                                                  will skip the first two lines and then extract column
     *                                                  information from the third line. columns will be ignored if
     *                                                  this is set to true.
     * @param int                $skipHeaderRows        If this is set, the task will skip the first skipHeaderRows
     *                                                  rows.
     */
    public function tsvFormat(
        ?array $columns = null,
        ?string $delimiter = null,
        ?string $listDelimiter = null,
        ?bool $findColumnsFromHeader = null,
        int $skipHeaderRows = 0
    ): self {
        $this->inputFormat = new TsvInputFormat(
            $columns,
            $delimiter,
            $listDelimiter,
            $findColumnsFromHeader,
            $skipHeaderRows
        );

        return $this;
    }

    /**
     * Specify that we use ORC as input format.
     *
     * To use the ORC input format, load the Druid Orc extension ( druid-orc-extensions).
     *
     * @param FlattenSpec|null $flattenSpec    Specifies flattening configuration for nested ORC data. See flattenSpec
     *                                         for more info.
     * @param bool|null        $binaryAsString Specifies if the binary orc column which is not logically marked as a
     *                                         string should be treated as a UTF-8 encoded string. Default is false.
     */
    public function orcFormat(?FlattenSpec $flattenSpec = null, ?bool $binaryAsString = null): self
    {
        $this->inputFormat = new OrcInputFormat($flattenSpec, $binaryAsString);

        return $this;
    }

    /**
     * Specify that we use Parquet as input format.
     *
     * To use the Parquet input format load the Druid Parquet extension (druid-parquet-extensions).
     *
     * @param FlattenSpec|null $flattenSpec    Define a flattenSpec to extract nested values from a Parquet file. Note
     *                                         that only 'path' expression are supported ('jq' is unavailable).
     * @param bool|null        $binaryAsString Specifies if the bytes parquet column which is not logically marked as a
     *                                         string or enum type should be treated as a UTF-8 encoded string.
     */
    public function parquetFormat(?FlattenSpec $flattenSpec = null, ?bool $binaryAsString = null): self
    {
        $this->inputFormat = new ParquetInputFormat($flattenSpec, $binaryAsString);

        return $this;
    }

    /**
     * Specify that we use Protobuf as input format.
     *
     * You need to include the druid-protobuf-extensions as an extension to use the Protobuf input format.
     *
     * @param array<string,string> $protoBytesDecoder Specifies how to decode bytes to Protobuf record. See below for
     *                                                an example.
     * @param FlattenSpec|null     $flattenSpec       Define a flattenSpec to extract nested values from a Parquet
     *                                                file. Note that only 'path' expression are supported ('jq' is
     *                                                unavailable).
     *
     * Example $protoBytesDecoder value:
     * ```
     * [
     *     "type" => "file",
     *     "descriptor" => "file:///tmp/metrics.desc",
     *     "protoMessageType" => "Metrics"
     * ]
     * ```
     *
     * @see https://druid.apache.org/docs/latest/ingestion/data-formats.html#protobuf
     */
    public function protobufFormat(array $protoBytesDecoder, ?FlattenSpec $flattenSpec = null): self
    {
        $this->inputFormat = new ProtobufInputFormat($protoBytesDecoder, $flattenSpec);

        return $this;
    }

    /**
     * Specify that we use Avro stream as input format.
     *
     * To use the Avro stream input format, load the Druid Avro extension (druid-avro-extensions).
     *
     * @param array<string,mixed> $avroBytesDecoder    Specifies how to decode bytes into an Avro record. The shape
     *                                                  depends on the decoder type (schema_registry, schema_inline,
     *                                                  schema_repo, etc.).
     * @param FlattenSpec|null    $flattenSpec         Specifies flattening configuration for nested Avro data.
     * @param bool|null           $binaryAsString      Treat binary Avro columns as UTF-8 strings. Default false.
     * @param bool|null           $extractUnionsByType Extract Avro union fields as a structured object keyed by the
     *                                                  union member type instead of the raw value. Default false.
     *
     * @see https://druid.apache.org/docs/latest/ingestion/data-formats#avro-stream
     */
    public function avroStreamFormat(
        array $avroBytesDecoder,
        ?FlattenSpec $flattenSpec = null,
        ?bool $binaryAsString = null,
        ?bool $extractUnionsByType = null
    ): self {
        $this->inputFormat = new AvroStreamInputFormat(
            $avroBytesDecoder,
            $flattenSpec,
            $binaryAsString,
            $extractUnionsByType
        );

        return $this;
    }

    /**
     * Specify that we use Avro Object Container Files as input format (batch).
     *
     * To use the Avro OCF input format, load the Druid Avro extension (druid-avro-extensions).
     *
     * @param FlattenSpec|null         $flattenSpec         Specifies flattening configuration for nested Avro data.
     * @param array<string,mixed>|null $schema              Optional reader schema as an Avro JSON record. When omitted
     *                                                       the writer schema embedded in the OCF file is used.
     * @param bool|null                $binaryAsString      Treat binary Avro columns as UTF-8 strings. Default false.
     * @param bool|null                $extractUnionsByType Extract Avro union fields as a structured object keyed by
     *                                                       the union member type instead of the raw value. Default
     *                                                       false.
     *
     * @see https://druid.apache.org/docs/latest/ingestion/data-formats#avro-ocf
     */
    public function avroOcfFormat(
        ?FlattenSpec $flattenSpec = null,
        ?array $schema = null,
        ?bool $binaryAsString = null,
        ?bool $extractUnionsByType = null
    ): self {
        $this->inputFormat = new AvroOcfInputFormat(
            $flattenSpec,
            $schema,
            $binaryAsString,
            $extractUnionsByType
        );

        return $this;
    }

    /**
     * Specify that we use the Kafka input format. This wraps another input format and exposes Kafka envelope
     * metadata (timestamp, headers, key, topic) as columns alongside the parsed value.
     *
     * @param InputFormatInterface      $valueFormat         Input format used to parse the Kafka record value.
     * @param InputFormatInterface|null $keyFormat           Input format used to parse the Kafka record key.
     * @param array<string,string>|null $headerFormat        Header decoder spec, e.g. ['type' => 'string', 'encoding'
     *                                                         => 'UTF-8'].
     * @param string|null               $headerColumnPrefix  Prefix for header columns. Druid default: 'kafka.header.'.
     * @param string|null               $keyColumnName       Column name for the parsed key. Druid default: 'kafka.key'.
     * @param string|null               $timestampColumnName Column name for the Kafka timestamp. Druid default:
     *                                                         'kafka.timestamp'.
     * @param string|null               $topicColumnName     Column name for the Kafka topic. Druid default:
     *                                                         'kafka.topic'.
     *
     * @see https://druid.apache.org/docs/latest/ingestion/data-formats#kafka
     */
    public function kafkaFormat(
        InputFormatInterface $valueFormat,
        ?InputFormatInterface $keyFormat = null,
        ?array $headerFormat = null,
        ?string $headerColumnPrefix = null,
        ?string $keyColumnName = null,
        ?string $timestampColumnName = null,
        ?string $topicColumnName = null
    ): self {
        $this->inputFormat = new KafkaInputFormat(
            $valueFormat,
            $keyFormat,
            $headerFormat,
            $headerColumnPrefix,
            $keyColumnName,
            $timestampColumnName,
            $topicColumnName
        );

        return $this;
    }
}

1	<?php
2	declare(strict_types=1);
3
4	namespace Level23\Druid\Concerns;
5
6	use Level23\Druid\InputFormats\FlattenSpec;
7	use Level23\Druid\InputFormats\CsvInputFormat;
8	use Level23\Druid\InputFormats\TsvInputFormat;
9	use Level23\Druid\InputFormats\OrcInputFormat;
10	use Level23\Druid\InputFormats\JsonInputFormat;
11	use Level23\Druid\InputFormats\KafkaInputFormat;
12	use Level23\Druid\InputFormats\AvroOcfInputFormat;
13	use Level23\Druid\InputFormats\ParquetInputFormat;
14	use Level23\Druid\InputFormats\ProtobufInputFormat;
15	use Level23\Druid\InputFormats\AvroStreamInputFormat;
16	use Level23\Druid\InputFormats\InputFormatInterface;
17
18	trait HasInputFormat
19	{
20	/**
21	* @var \Level23\Druid\InputFormats\InputFormatInterface\|null
22	*/
23	protected ?InputFormatInterface $inputFormat = null;
24
25	/**
26	* Specify that we use JSON as input format.
27	*
28	* @param FlattenSpec\|null $flattenSpec Specifies flattening configuration for nested JSON data. See
29	* flattenSpec for more info.
30	* @param array<string,bool>\|null $features JSON parser features supported by Jackson library. Those features
31	* will be applied when parsing the input JSON data.
32	*
33	* @see https://github.com/FasterXML/jackson-core/wiki/JsonParser-Features
34	*/
35	public function jsonFormat(?FlattenSpec $flattenSpec = null, ?array $features = null): self
36	{
37	$this->inputFormat = new JsonInputFormat($flattenSpec, $features);	3✔
38
39	return $this;	3✔
40	}
41
42	/**
43	* Specify that we use CSV as input format.
44	*
45	* @param string[]\|null $columns Specifies the columns of the data. The columns should be in the same
46	* order with the columns of your data.
47	* @param string\|null $listDelimiter A custom delimiter for multi-value dimensions.
48	* @param bool\|null $findColumnsFromHeader If this is set, the task will find the column names from the header
49	* row. Note that skipHeaderRows will be applied before finding column
50	* names from the header. For example, if you set skipHeaderRows to 2
51	* and findColumnsFromHeader to true, the task will skip the first two
52	* lines and then extract column information from the third line.
53	* columns will be ignored if this is set to true.
54	* @param int $skipHeaderRows If this is set, the task will skip the first skipHeaderRows rows.
55	*/
56	public function csvFormat(
57	?array $columns = null,
58	?string $listDelimiter = null,
59	?bool $findColumnsFromHeader = null,
60	int $skipHeaderRows = 0
61	): self {
62	$this->inputFormat = new CsvInputFormat($columns, $listDelimiter, $findColumnsFromHeader, $skipHeaderRows);	3✔
63
64	return $this;	3✔
65	}
66
67	/**
68	* Specify that we use TSV as input format.
69	*
70	* @param array<string>\|null $columns Specifies the columns of the data. The columns should be in the
71	* same order with the columns of your data.
72	* @param string\|null $delimiter A custom delimiter for data values.
73	* @param string\|null $listDelimiter A custom delimiter for multi-value dimensions.
74	* @param bool\|null $findColumnsFromHeader If this is set, the task will find the column names from the
75	* header row. Note that skipHeaderRows will be applied before
76	* finding column names from the header. For example, if you set
77	* skipHeaderRows to 2 and findColumnsFromHeader to true, the task
78	* will skip the first two lines and then extract column
79	* information from the third line. columns will be ignored if
80	* this is set to true.
81	* @param int $skipHeaderRows If this is set, the task will skip the first skipHeaderRows
82	* rows.
83	*/
84	public function tsvFormat(
85	?array $columns = null,
86	?string $delimiter = null,
87	?string $listDelimiter = null,
88	?bool $findColumnsFromHeader = null,
89	int $skipHeaderRows = 0
90	): self {
91	$this->inputFormat = new TsvInputFormat(	3✔
92	$columns,	3✔
93	$delimiter,	3✔
94	$listDelimiter,	3✔
95	$findColumnsFromHeader,	3✔
96	$skipHeaderRows	3✔
97	);	3✔
98
99	return $this;	3✔
100	}
101
102	/**
103	* Specify that we use ORC as input format.
104	*
105	* To use the ORC input format, load the Druid Orc extension ( druid-orc-extensions).
106	*
107	* @param FlattenSpec\|null $flattenSpec Specifies flattening configuration for nested ORC data. See flattenSpec
108	* for more info.
109	* @param bool\|null $binaryAsString Specifies if the binary orc column which is not logically marked as a
110	* string should be treated as a UTF-8 encoded string. Default is false.
111	*/
112	public function orcFormat(?FlattenSpec $flattenSpec = null, ?bool $binaryAsString = null): self
113	{
114	$this->inputFormat = new OrcInputFormat($flattenSpec, $binaryAsString);	3✔
115
116	return $this;	3✔
117	}
118
119	/**
120	* Specify that we use Parquet as input format.
121	*
122	* To use the Parquet input format load the Druid Parquet extension (druid-parquet-extensions).
123	*
124	* @param FlattenSpec\|null $flattenSpec Define a flattenSpec to extract nested values from a Parquet file. Note
125	* that only 'path' expression are supported ('jq' is unavailable).
126	* @param bool\|null $binaryAsString Specifies if the bytes parquet column which is not logically marked as a
127	* string or enum type should be treated as a UTF-8 encoded string.
128	*/
129	public function parquetFormat(?FlattenSpec $flattenSpec = null, ?bool $binaryAsString = null): self
130	{
131	$this->inputFormat = new ParquetInputFormat($flattenSpec, $binaryAsString);	3✔
132
133	return $this;	3✔
134	}
135
136	/**
137	* Specify that we use Protobuf as input format.
138	*
139	* You need to include the druid-protobuf-extensions as an extension to use the Protobuf input format.
140	*
141	* @param array<string,string> $protoBytesDecoder Specifies how to decode bytes to Protobuf record. See below for
142	* an example.
143	* @param FlattenSpec\|null $flattenSpec Define a flattenSpec to extract nested values from a Parquet
144	* file. Note that only 'path' expression are supported ('jq' is
145	* unavailable).
146	*
147	* Example $protoBytesDecoder value:
148	* ```
149	* [
150	* "type" => "file",
151	* "descriptor" => "file:///tmp/metrics.desc",
152	* "protoMessageType" => "Metrics"
153	* ]
154	* ```
155	*
156	* @see https://druid.apache.org/docs/latest/ingestion/data-formats.html#protobuf
157	*/
158	public function protobufFormat(array $protoBytesDecoder, ?FlattenSpec $flattenSpec = null): self
159	{
160	$this->inputFormat = new ProtobufInputFormat($protoBytesDecoder, $flattenSpec);	3✔
161
162	return $this;	3✔
163	}
164
165	/**
166	* Specify that we use Avro stream as input format.
167	*
168	* To use the Avro stream input format, load the Druid Avro extension (druid-avro-extensions).
169	*
170	* @param array<string,mixed> $avroBytesDecoder Specifies how to decode bytes into an Avro record. The shape
171	* depends on the decoder type (schema_registry, schema_inline,
172	* schema_repo, etc.).
173	* @param FlattenSpec\|null $flattenSpec Specifies flattening configuration for nested Avro data.
174	* @param bool\|null $binaryAsString Treat binary Avro columns as UTF-8 strings. Default false.
175	* @param bool\|null $extractUnionsByType Extract Avro union fields as a structured object keyed by the
176	* union member type instead of the raw value. Default false.
177	*
178	* @see https://druid.apache.org/docs/latest/ingestion/data-formats#avro-stream
179	*/
180	public function avroStreamFormat(
181	array $avroBytesDecoder,
182	?FlattenSpec $flattenSpec = null,
183	?bool $binaryAsString = null,
184	?bool $extractUnionsByType = null
185	): self {
NEW 186	$this->inputFormat = new AvroStreamInputFormat(	×
NEW 187	$avroBytesDecoder,	×
NEW 188	$flattenSpec,	×
NEW 189	$binaryAsString,	×
NEW 190	$extractUnionsByType	×
NEW 191	);	×
192
NEW 193	return $this;	×
194	}
195
196	/**
197	* Specify that we use Avro Object Container Files as input format (batch).
198	*
199	* To use the Avro OCF input format, load the Druid Avro extension (druid-avro-extensions).
200	*
201	* @param FlattenSpec\|null $flattenSpec Specifies flattening configuration for nested Avro data.
202	* @param array<string,mixed>\|null $schema Optional reader schema as an Avro JSON record. When omitted
203	* the writer schema embedded in the OCF file is used.
204	* @param bool\|null $binaryAsString Treat binary Avro columns as UTF-8 strings. Default false.
205	* @param bool\|null $extractUnionsByType Extract Avro union fields as a structured object keyed by
206	* the union member type instead of the raw value. Default
207	* false.
208	*
209	* @see https://druid.apache.org/docs/latest/ingestion/data-formats#avro-ocf
210	*/
211	public function avroOcfFormat(
212	?FlattenSpec $flattenSpec = null,
213	?array $schema = null,
214	?bool $binaryAsString = null,
215	?bool $extractUnionsByType = null
216	): self {
NEW 217	$this->inputFormat = new AvroOcfInputFormat(	×
NEW 218	$flattenSpec,	×
NEW 219	$schema,	×
NEW 220	$binaryAsString,	×
NEW 221	$extractUnionsByType	×
NEW 222	);	×
223
NEW 224	return $this;	×
225	}
226
227	/**
228	* Specify that we use the Kafka input format. This wraps another input format and exposes Kafka envelope
229	* metadata (timestamp, headers, key, topic) as columns alongside the parsed value.
230	*
231	* @param InputFormatInterface $valueFormat Input format used to parse the Kafka record value.
232	* @param InputFormatInterface\|null $keyFormat Input format used to parse the Kafka record key.
233	* @param array<string,string>\|null $headerFormat Header decoder spec, e.g. ['type' => 'string', 'encoding'
234	* => 'UTF-8'].
235	* @param string\|null $headerColumnPrefix Prefix for header columns. Druid default: 'kafka.header.'.
236	* @param string\|null $keyColumnName Column name for the parsed key. Druid default: 'kafka.key'.
237	* @param string\|null $timestampColumnName Column name for the Kafka timestamp. Druid default:
238	* 'kafka.timestamp'.
239	* @param string\|null $topicColumnName Column name for the Kafka topic. Druid default:
240	* 'kafka.topic'.
241	*
242	* @see https://druid.apache.org/docs/latest/ingestion/data-formats#kafka
243	*/
244	public function kafkaFormat(
245	InputFormatInterface $valueFormat,
246	?InputFormatInterface $keyFormat = null,
247	?array $headerFormat = null,
248	?string $headerColumnPrefix = null,
249	?string $keyColumnName = null,
250	?string $timestampColumnName = null,
251	?string $topicColumnName = null
252	): self {
NEW 253	$this->inputFormat = new KafkaInputFormat(	×
NEW 254	$valueFormat,	×
NEW 255	$keyFormat,	×
NEW 256	$headerFormat,	×
NEW 257	$headerColumnPrefix,	×
NEW 258	$keyColumnName,	×
NEW 259	$timestampColumnName,	×
NEW 260	$topicColumnName	×
NEW 261	);	×
262
NEW 263	return $this;	×
264	}
265	}

level23 / druid-client / 27420372206

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous