Skip to content

Commit

Permalink
Checkpoint on adding Daffodil to Drill
Browse files Browse the repository at this point in the history
Depends on a mbeckerle-specific 3.7.0-SNAPSHOT fork of
Daffodil.

New format-daffodil module created

A basic unit test of DFDL schema with a single int field
works in that drill metadata is created by traversing the
schema, then data is populated using drill's row-set-writer
being drivn by a Daffodil InfosetOutputter from a
Daffodil parse.

Still uses absolute paths for the schemaFileURI.
(which is cheating. Wouldn't work in a true distributed
drill environment.)

We have yet to work out how to enable Drill to provide
access for DFDL schemas in XML form with include/import
to be resolved.

The input data stream is, however, being accessed in the
proper Drill manner. Gunzip happened automatically. Nice.

Note: Fix boxed Boolean vs. boolean problem. Don't use
boxed primitives in Format config objects.

https://github.com/apache/drill/issues/2835ssues/2835
  • Loading branch information
mbeckerle committed Nov 5, 2023
1 parent aa52f05 commit eb418bf
Show file tree
Hide file tree
Showing 26 changed files with 1,659 additions and 1 deletion.
2 changes: 2 additions & 0 deletions contrib/format-daffodil/.gitignore
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
# Directory to store oauth tokens for testing Googlesheets Storage plugin
/src/test/resources/logback-test.xml
41 changes: 41 additions & 0 deletions contrib/format-daffodil/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,41 @@
# Daffodil 'Format' Reader
This plugin enables Drill to read DFDL-described data by way of the Apache Daffodil DFDL implementation.

## Configuration
* ?? src/main/resources/bootstrap-format-plugins.json
* ?? src/main/resources/drill-module.conf

TBD - how to enable/disable validation by Daffodil?
Note that this should be per sql query that reads data.

Note that schemas should be able to be loaded once in the config file and then referenced from queries.

The query should include whether to validate (daffodil 'limited' is the only option) or not.

## Data Types

TBD

## Provided Schema
The Daffodil Format Reader supports pre-compiled DFDL schemas.

TBD example query, data, and schema
```sql
SELECT * FROM dfs.`/tmp/data.dat`(type => 'daffodil',
schema => "schema_identifier_from_config");
```

## Limitations: TBD

This version of the daffodil interface only supports data querying, there is no ability to output (aka unparse in DFDL terminology).

The data is parsed fully from its native form into a Drill data structure held in memory.
No attempt is made to avoid access to parts of the DFDL-described data that are not needed to answer the query.

If the data is not well-formed, an error occurs and the query fails.

If the data is invalid, and validity checking by Daffodil is enabled, then an error occurs and the query fails.

## Future Functionality

* TBD
94 changes: 94 additions & 0 deletions contrib/format-daffodil/pom.xml
Original file line number Diff line number Diff line change
@@ -0,0 +1,94 @@
<?xml version="1.0"?>
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>

<parent>
<artifactId>drill-contrib-parent</artifactId>
<groupId>org.apache.drill.contrib</groupId>
<version>1.22.0-SNAPSHOT</version>
</parent>

<artifactId>drill-format-daffodil</artifactId>
<name>Drill : Contrib : Format : Daffodil</name>

<dependencies>
<dependency>
<groupId>org.apache.drill.exec</groupId>
<artifactId>drill-java-exec</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.apache.daffodil</groupId>
<artifactId>daffodil-japi_2.12</artifactId>
<version>3.7.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.daffodil</groupId>
<artifactId>daffodil-runtime1_2.12</artifactId>
<version>3.7.0-SNAPSHOT</version>
</dependency>
<!-- Test dependencies -->
<dependency>
<groupId>org.apache.drill.exec</groupId>
<artifactId>drill-java-exec</artifactId>
<classifier>tests</classifier>
<version>${project.version}</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>org.apache.drill</groupId>
<artifactId>drill-common</artifactId>
<classifier>tests</classifier>
<version>${project.version}</version>
<scope>test</scope>
</dependency>
</dependencies>

<build>
<plugins>
<plugin>
<artifactId>maven-resources-plugin</artifactId>
<executions>
<execution>
<id>copy-java-sources</id>
<phase>process-sources</phase>
<goals>
<goal>copy-resources</goal>
</goals>
<configuration>
<outputDirectory>${basedir}/target/classes/org/apache/drill/exec/store/daffodil
</outputDirectory>
<resources>
<resource>
<directory>src/main/java/org/apache/drill/exec/store/daffodil</directory>
<filtering>true</filtering>
</resource>
</resources>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
Original file line number Diff line number Diff line change
@@ -0,0 +1,184 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.drill.exec.store.daffodil;

import java.io.InputStream;
import java.net.URI;
import java.net.URISyntaxException;

import org.apache.daffodil.japi.DataProcessor;
import org.apache.drill.common.AutoCloseables;
import org.apache.drill.common.exceptions.CustomErrorContext;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.physical.impl.scan.v3.ManagedReader;
import org.apache.drill.exec.physical.impl.scan.v3.file.FileDescrip;
import org.apache.drill.exec.physical.impl.scan.v3.file.FileSchemaNegotiator;
import org.apache.drill.exec.physical.resultSet.RowSetLoader;
import org.apache.drill.exec.record.metadata.TupleMetadata;
import org.apache.drill.exec.store.daffodil.schema.DaffodilDataProcessorFactory;
import org.apache.drill.exec.store.dfs.DrillFileSystem;
import org.apache.drill.exec.store.dfs.easy.EasySubScan;
import org.apache.hadoop.fs.Path;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.apache.drill.exec.store.daffodil.schema.DrillDaffodilSchemaUtils.daffodilDataProcessorToDrillSchema;


public class DaffodilBatchReader implements ManagedReader {

private static final Logger logger = LoggerFactory.getLogger(DaffodilBatchReader.class);
private final DaffodilFormatConfig dafConfig;
private final RowSetLoader rowSetLoader;
private final CustomErrorContext errorContext;
private final DaffodilMessageParser dafParser;
private final InputStream dataInputStream;

static class DaffodilReaderConfig {
final DaffodilFormatPlugin plugin;
DaffodilReaderConfig(DaffodilFormatPlugin plugin) {
this.plugin = plugin;
}
}

public DaffodilBatchReader (DaffodilReaderConfig readerConfig, EasySubScan scan, FileSchemaNegotiator negotiator) {

errorContext = negotiator.parentErrorContext();
this.dafConfig = readerConfig.plugin.getConfig();

String schemaURIString = dafConfig.getSchemaURI(); // "schema/complexArray1.dfdl.xsd";
String rootName = dafConfig.getRootName();
String rootNamespace = dafConfig.getRootNamespace();
Boolean validationMode = dafConfig.getValidationMode();

URI dfdlSchemaURI;
try {
dfdlSchemaURI = new URI(schemaURIString);
} catch (URISyntaxException e) {
throw UserException.validationError(e)
.build(logger);
}

FileDescrip file = negotiator.file();
DrillFileSystem fs = file.fileSystem();
URI fsSchemaURI = fs.getUri().resolve(dfdlSchemaURI);


DaffodilDataProcessorFactory dpf = new DaffodilDataProcessorFactory();
DataProcessor dp;
try {
dp = dpf.getDataProcessor(fsSchemaURI, validationMode, rootName, rootNamespace);
} catch (Exception e) {
throw UserException.dataReadError(e)
.message(String.format("Failed to get Daffodil DFDL processor for: %s", fsSchemaURI))
.addContext(errorContext).addContext(e.getMessage()).build(logger);
}
// Create the corresponding Drill schema.
// Note: this could be a very large schema. Think of a large complex RDBMS schema,
// all of it, hundreds of tables, but all part of the same metadata tree.
TupleMetadata drillSchema = daffodilDataProcessorToDrillSchema(dp);
// Inform Drill about the schema
negotiator.tableSchema(drillSchema, true);

//
// DATA TIME: Next we construct the runtime objects, and open files.
//
// We get the DaffodilMessageParser, which is a stateful driver for daffodil that
// actually does the parsing.
rowSetLoader = negotiator.build().writer();

// We construct the Daffodil InfosetOutputter which the daffodil parser uses to
// convert infoset event calls to fill in a Drill row via a rowSetLoader.
DaffodilDrillInfosetOutputter outputter = new DaffodilDrillInfosetOutputter(rowSetLoader);

// Now we can setup the dafParser with the outputter it will drive with
// the parser-produced infoset.
dafParser = new DaffodilMessageParser(dp); // needs further initialization after this.
dafParser.setInfosetOutputter(outputter);

Path dataPath = file.split().getPath();
// Lastly, we open the data stream
try {
dataInputStream = fs.openPossiblyCompressedStream(dataPath);
} catch (Exception e) {
throw UserException.dataReadError(e)
.message(String.format("Failed to open input file: %s", dataPath.toString()))
.addContext(errorContext).addContext(e.getMessage()).build(logger);
}
// And lastly,... tell daffodil the input data stream.
dafParser.setInputStream(dataInputStream);
}


/**
* This is the core of actual processing - data movement from Daffodil to Drill.
*
* If there is space in the batch, and there is data available to parse
* then this calls the daffodil parser, which parses data, delivering it to the rowWriter
* by way of the infoset outputter.
* <p>
* Repeats until the rowWriter is full (a batch is full), or there is no more data, or
* a parse error ends execution with a throw.
* <p>
* Validation errors and other warnings are not errors and are logged but do not cause
* parsing to fail/throw.
* @return true if there are rows retrieved, false if no rows were retrieved, which means
* no more will ever be retrieved (end of data).
* @throws RuntimeException on parse errors.
*/
@Override
public boolean next() {
// Check assumed invariants
// We don't know if there is data or not. This could be called on an empty data file.
// We DO know that this won't be called if there is no space in the batch for even 1
// row.
if (dafParser.isEOF()) {
return false; // return without even checking for more rows or trying to parse.
}
while (rowSetLoader.start() && !dafParser.isEOF()) { // we never zero-trip this loop.
// the predicate is always true once.
try {
dafParser.parse();
if (dafParser.isProcessingError()) {
throw UserException.dataReadError().message(dafParser.getDiagnosticsAsString())
.addContext(errorContext).build(logger);
}
if (dafParser.isValidationError()) {
logger.warn(dafParser.getDiagnosticsAsString());
// Note that even if daffodil is set to not validate, validation errors may still occur
// from DFDL's "recoverableError" assertions.
}
} catch (Exception e) {
throw UserException.dataReadError(e).message("Error parsing file: " + e.getMessage())
.addContext(errorContext).build(logger);
}
// success! We have parsed data and populated a row.
rowSetLoader.save();
}
int nRows = rowSetLoader.rowCount();
assert nRows > 0; // This cannot be zero. If the parse failed we will have already thrown out of here.
return true;
}

@Override
public void close() {
if (dataInputStream != null)
AutoCloseables.closeSilently(dataInputStream);
}
}
Loading

0 comments on commit eb418bf

Please sign in to comment.