| Commit message (Collapse) | Author | Age | Files | Lines |
|---|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
JIRA: https://issues.apache.org/jira/browse/SPARK-13674
## What changes were proposed in this pull request?
Sample operator doesn't support wholestage codegen now. This pr is to add support to it.
## How was this patch tested?
A test is added into `BenchmarkWholeStageCodegen`. Besides, all tests should be passed.
Author: Liang-Chi Hsieh <simonh@tw.ibm.com>
Author: Liang-Chi Hsieh <viirya@gmail.com>
Closes #11517 from viirya/add-wholestage-sample.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR adds the function `window` as a column expression.
`window` can be used to bucket rows into time windows given a time column. With this expression, performing time series analysis on batch data, as well as streaming data should become much more simpler.
### Usage
Assume the following schema:
`sensor_id, measurement, timestamp`
To average 5 minute data every 1 minute (window length of 5 minutes, slide duration of 1 minute), we will use:
```scala
df.groupBy(window("timestamp", “5 minutes”, “1 minute”), "sensor_id")
.agg(mean("measurement").as("avg_meas"))
```
This will generate windows such as:
```
09:00:00-09:05:00
09:01:00-09:06:00
09:02:00-09:07:00 ...
```
Intervals will start at every `slideDuration` starting at the unix epoch (1970-01-01 00:00:00 UTC).
To start intervals at a different point of time, e.g. 30 seconds after a minute, the `startTime` parameter can be used.
```scala
df.groupBy(window("timestamp", “5 minutes”, “1 minute”, "30 second"), "sensor_id")
.agg(mean("measurement").as("avg_meas"))
```
This will generate windows such as:
```
09:00:30-09:05:30
09:01:30-09:06:30
09:02:30-09:07:30 ...
```
Support for Python will be made in a follow up PR after this.
## How was this patch tested?
This patch has some basic unit tests for the `TimeWindow` expression testing that the parameters pass validation, and it also has some unit/integration tests testing the correctness of the windowing and usability in complex operations (multi-column grouping, multi-column projections, joins).
Author: Burak Yavuz <brkyvz@gmail.com>
Author: Michael Armbrust <michael@databricks.com>
Closes #12008 from brkyvz/df-time-window.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This patch enables use of OrcRelation for SQL queries which read data from Hive tables. Changes in this patch:
- Added a new rule `OrcConversions` which would alter the plan to use `OrcRelation`. In this diff, the conversion is done only for reads.
- Added a new config `spark.sql.hive.convertMetastoreOrc` to control the conversion
BEFORE
```
scala> hqlContext.sql("SELECT * FROM orc_table").explain(true)
== Parsed Logical Plan ==
'Project [unresolvedalias(*, None)]
+- 'UnresolvedRelation `orc_table`, None
== Analyzed Logical Plan ==
key: string, value: string
Project [key#171,value#172]
+- MetastoreRelation default, orc_table, None
== Optimized Logical Plan ==
MetastoreRelation default, orc_table, None
== Physical Plan ==
HiveTableScan [key#171,value#172], MetastoreRelation default, orc_table, None
```
AFTER
```
scala> hqlContext.sql("SELECT * FROM orc_table").explain(true)
== Parsed Logical Plan ==
'Project [unresolvedalias(*, None)]
+- 'UnresolvedRelation `orc_table`, None
== Analyzed Logical Plan ==
key: string, value: string
Project [key#76,value#77]
+- SubqueryAlias orc_table
+- Relation[key#76,value#77] ORC part: struct<>, data: struct<key:string,value:string>
== Optimized Logical Plan ==
Relation[key#76,value#77] ORC part: struct<>, data: struct<key:string,value:string>
== Physical Plan ==
WholeStageCodegen
: +- Scan ORC part: struct<>, data: struct<key:string,value:string>[key#76,value#77] InputPaths: file:/user/hive/warehouse/orc_table
```
## How was this patch tested?
- Added a new unit test. Ran existing unit tests
- Ran with production like data
## Performance gains
Ran on a production table in Facebook (note that the data was in DWRF file format which is similar to ORC)
Best case : when there was no matching rows for the predicate in the query (everything is filtered out)
```
CPU time Wall time Total wall time across all tasks
================================================================
Without the change 541_515 sec 25.0 mins 165.8 hours
With change 407 sec 1.5 mins 15 mins
```
Average case: A subset of rows in the data match the query predicate
```
CPU time Wall time Total wall time across all tasks
================================================================
Without the change 624_630 sec 31.0 mins 199.0 h
With change 14_769 sec 5.3 mins 7.7 h
```
Author: Tejas Patil <tejasp@fb.com>
Closes #11891 from tejasapatil/orc_ppd.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
`Expand` operator now uses its child plan's constraints as its valid constraints (i.e., the base of constraints). This is not correct because `Expand` will set its group by attributes to null values. So the nullability of these attributes should be true.
E.g., for an `Expand` operator like:
val input = LocalRelation('a.int, 'b.int, 'c.int).where('c.attr > 10 && 'a.attr < 5 && 'b.attr > 2)
Expand(
Seq(
Seq('c, Literal.create(null, StringType), 1),
Seq('c, 'a, 2)),
Seq('c, 'a, 'gid.int),
Project(Seq('a, 'c), input))
The `Project` operator has the constraints `IsNotNull('a)`, `IsNotNull('b)` and `IsNotNull('c)`. But the `Expand` should not have `IsNotNull('a)` in its constraints.
This PR is the first step for this issue and remove invalid constraints of `Expand` operator.
A test is added to `ConstraintPropagationSuite`.
Author: Liang-Chi Hsieh <simonh@tw.ibm.com>
Author: Michael Armbrust <michael@databricks.com>
Closes #11995 from viirya/fix-expand-constraints.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
JIRA: https://issues.apache.org/jira/browse/SPARK-13995
We infer relative `IsNotNull` constraints from logical plan's expressions in `constructIsNotNullConstraints` now. However, we don't consider the case of (nested) `Cast`.
For example:
val tr = LocalRelation('a.int, 'b.long)
val plan = tr.where('a.attr === 'b.attr).analyze
Then, the plan's constraints will have `IsNotNull(Cast(resolveColumn(tr, "a"), LongType))`, instead of `IsNotNull(resolveColumn(tr, "a"))`. This PR fixes it.
Besides, as `IsNotNull` constraints are most useful for `Attribute`, we should do recursing through any `Expression` that is null intolerant and construct `IsNotNull` constraints for all `Attribute`s under these Expressions.
For example, consider the following constraints:
val df = Seq((1,2,3)).toDF("a", "b", "c")
df.where("a + b = c").queryExecution.analyzed.constraints
The inferred isnotnull constraints should be isnotnull(a), isnotnull(b), isnotnull(c), instead of isnotnull(a + c) and isnotnull(c).
## How was this patch tested?
Test is added into `ConstraintPropagationSuite`.
Author: Liang-Chi Hsieh <simonh@tw.ibm.com>
Closes #11809 from viirya/constraint-cast.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
`debugCodegen`
## What changes were proposed in this pull request?
This PR updates the usage comments of `debug` according to the following commits.
- [SPARK-9754](https://issues.apache.org/jira/browse/SPARK-9754) removed `typeCheck`.
- [SPARK-14227](https://issues.apache.org/jira/browse/SPARK-14227) added `debugCodegen`.
## How was this patch tested?
Manual.
Author: Dongjoon Hyun <dongjoon@apache.org>
Closes #12094 from dongjoon-hyun/minor_fix_debug_usage.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
, and lock/unlock operations.
## What changes were proposed in this pull request?
This PR throws Unsupported Operation exception for create index, drop index, alter index , lock table , lock database, unlock table, and unlock database operations that are not supported in Spark SQL. Currently these operations are executed executed by Hive.
Error:
spark-sql> drop index my_index on my_table;
Error in query:
Unsupported operation: drop index(line 1, pos 0)
## How was this patch tested?
Added test cases to HiveQuerySuite
yhuai hvanhovell andrewor14
Author: sureshthalamati <suresh.thalamati@gmail.com>
Closes #12069 from sureshthalamati/unsupported_ddl_spark-14133.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
SHOW TABLE to use table identifier pattern
## What changes were proposed in this pull request?
This PR addresses the following
1. Supports native execution of SHOW DATABASES command
2. Fixes SHOW TABLES to apply the identifier_with_wildcards pattern if supplied.
SHOW TABLE syntax
```
SHOW TABLES [IN database_name] ['identifier_with_wildcards'];
```
SHOW DATABASES syntax
```
SHOW (DATABASES|SCHEMAS) [LIKE 'identifier_with_wildcards'];
```
## How was this patch tested?
Tests added in SQLQuerySuite (both hive and sql contexts) and DDLCommandSuite
Note: Since the table name pattern was not working , tests are added in both SQLQuerySuite to
verify the application of the table pattern.
Author: Dilip Biswal <dbiswal@us.ibm.com>
Closes #11991 from dilipbiswal/dkb_show_database.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR implements `FileFormat.buildReader()` for the LibSVM data source. Besides that, a new interface method `prepareRead()` is added to `FileFormat`:
```scala
def prepareRead(
sqlContext: SQLContext,
options: Map[String, String],
files: Seq[FileStatus]): Map[String, String] = options
```
After migrating from `buildInternalScan()` to `buildReader()`, we lost the opportunity to collect necessary global information, since `buildReader()` works in a per-partition manner. For example, LibSVM needs to infer the total number of features if the `numFeatures` data source option is not set. Any necessary collected global information should be returned using the data source options map. By default, this method just returns the original options untouched.
An alternative approach is to absorb `inferSchema()` into `prepareRead()`, since schema inference is also some kind of global information gathering. However, this approach wasn't chosen because schema inference is optional, while `prepareRead()` must be called whenever a `HadoopFsRelation` based data source relation is instantiated.
One unaddressed problem is that, when `numFeatures` is absent, now the input data will be scanned twice. The `buildInternalScan()` code path doesn't need to do this because it caches the raw parsed RDD in memory before computing the total number of features. However, with `FileScanRDD`, the raw parsed RDD is created in a different way (e.g. partitioning) from the final RDD.
## How was this patch tested?
Tested using existing test suites.
Author: Cheng Lian <lian@databricks.com>
Closes #12088 from liancheng/spark-14295-libsvm-build-reader.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR support multiple Python UDFs within single batch, also improve the performance.
```python
>>> from pyspark.sql.types import IntegerType
>>> sqlContext.registerFunction("double", lambda x: x * 2, IntegerType())
>>> sqlContext.registerFunction("add", lambda x, y: x + y, IntegerType())
>>> sqlContext.sql("SELECT double(add(1, 2)), add(double(2), 1)").explain(True)
== Parsed Logical Plan ==
'Project [unresolvedalias('double('add(1, 2)), None),unresolvedalias('add('double(2), 1), None)]
+- OneRowRelation$
== Analyzed Logical Plan ==
double(add(1, 2)): int, add(double(2), 1): int
Project [double(add(1, 2))#14,add(double(2), 1)#15]
+- Project [double(add(1, 2))#14,add(double(2), 1)#15]
+- Project [pythonUDF0#16 AS double(add(1, 2))#14,pythonUDF0#18 AS add(double(2), 1)#15]
+- EvaluatePython [add(pythonUDF1#17, 1)], [pythonUDF0#18]
+- EvaluatePython [double(add(1, 2)),double(2)], [pythonUDF0#16,pythonUDF1#17]
+- OneRowRelation$
== Optimized Logical Plan ==
Project [pythonUDF0#16 AS double(add(1, 2))#14,pythonUDF0#18 AS add(double(2), 1)#15]
+- EvaluatePython [add(pythonUDF1#17, 1)], [pythonUDF0#18]
+- EvaluatePython [double(add(1, 2)),double(2)], [pythonUDF0#16,pythonUDF1#17]
+- OneRowRelation$
== Physical Plan ==
WholeStageCodegen
: +- Project [pythonUDF0#16 AS double(add(1, 2))#14,pythonUDF0#18 AS add(double(2), 1)#15]
: +- INPUT
+- !BatchPythonEvaluation [add(pythonUDF1#17, 1)], [pythonUDF0#16,pythonUDF1#17,pythonUDF0#18]
+- !BatchPythonEvaluation [double(add(1, 2)),double(2)], [pythonUDF0#16,pythonUDF1#17]
+- Scan OneRowRelation[]
```
## How was this patch tested?
Added new tests.
Using the following script to benchmark 1, 2 and 3 udfs,
```
df = sqlContext.range(1, 1 << 23, 1, 4)
double = F.udf(lambda x: x * 2, LongType())
print df.select(double(df.id)).count()
print df.select(double(df.id), double(df.id + 1)).count()
print df.select(double(df.id), double(df.id + 1), double(df.id + 2)).count()
```
Here is the results:
N | Before | After | speed up
---- |------------ | -------------|------
1 | 22 s | 7 s | 3.1X
2 | 38 s | 13 s | 2.9X
3 | 58 s | 16 s | 3.6X
This benchmark ran locally with 4 CPUs. For 3 UDFs, it launched 12 Python before before this patch, 4 process after this patch. After this patch, it will use less memory for multiple UDFs than before (less buffering).
Author: Davies Liu <davies@databricks.com>
Closes #12057 from davies/multi_udfs.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
`java.io.tmpdir` folder
## What changes were proposed in this pull request?
If I press `CTRL-C` when running these tests, the temp files will be left in `sql/core` folder and I need to delete them manually. It's annoying. This PR just moves the temp files to the `java.io.tmpdir` folder and add a name prefix for them.
## How was this patch tested?
Existing Jenkins tests
Author: Shixiong Zhu <shixiong@databricks.com>
Closes #12093 from zsxwing/temp-file.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
This PR is to provide native parsing support for DDL commands: `Alter View`. Since its AST trees are highly similar to `Alter Table`. Thus, both implementation are integrated into the same one.
Based on the Hive DDL document:
https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL and https://cwiki.apache.org/confluence/display/Hive/PartitionedViews
**Syntax:**
```SQL
ALTER VIEW view_name RENAME TO new_view_name
```
- to change the name of a view to a different name
**Syntax:**
```SQL
ALTER VIEW view_name SET TBLPROPERTIES ('comment' = new_comment);
```
- to add metadata to a view
**Syntax:**
```SQL
ALTER VIEW view_name UNSET TBLPROPERTIES [IF EXISTS] ('comment', 'key')
```
- to remove metadata from a view
**Syntax:**
```SQL
ALTER VIEW view_name ADD [IF NOT EXISTS] PARTITION spec1[, PARTITION spec2, ...]
```
- to add the partitioning metadata for a view.
- the syntax of partition spec in `ALTER VIEW` is identical to `ALTER TABLE`, **EXCEPT** that it is **ILLEGAL** to specify a `LOCATION` clause.
**Syntax:**
```SQL
ALTER VIEW view_name DROP [IF EXISTS] PARTITION spec1[, PARTITION spec2, ...]
```
- to drop the related partition metadata for a view.
Added the related test cases to `DDLCommandSuite`
Author: gatorsmile <gatorsmile@gmail.com>
Author: xiaoli <lixiao1983@gmail.com>
Author: Xiao Li <xiaoli@Xiaos-MacBook-Pro.local>
Closes #11987 from gatorsmile/parseAlterView.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
Fixes a minor bug in the record reader constructor that was possibly introduced during refactoring.
## How was this patch tested?
N/A
Author: Sameer Agarwal <sameer@databricks.com>
Closes #12070 from sameeragarwal/vectorized-rr.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR proposes a new data-structure based on a vectorized hashmap that can be potentially _codegened_ in `TungstenAggregate` to speed up aggregates with group by. Micro-benchmarks show a 10x improvement over the current `BytesToBytes` aggregation map.
## How was this patch tested?
Intel(R) Core(TM) i7-4960HQ CPU 2.60GHz
BytesToBytesMap: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative
-------------------------------------------------------------------------------------------
hash 108 / 119 96.9 10.3 1.0X
fast hash 63 / 70 166.2 6.0 1.7X
arrayEqual 70 / 73 150.8 6.6 1.6X
Java HashMap (Long) 141 / 200 74.3 13.5 0.8X
Java HashMap (two ints) 145 / 185 72.3 13.8 0.7X
Java HashMap (UnsafeRow) 499 / 524 21.0 47.6 0.2X
BytesToBytesMap (off Heap) 483 / 548 21.7 46.0 0.2X
BytesToBytesMap (on Heap) 485 / 562 21.6 46.2 0.2X
Vectorized Hashmap 54 / 60 193.7 5.2 2.0X
Author: Sameer Agarwal <sameer@databricks.com>
Closes #12055 from sameeragarwal/vectorized-hashmap.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
### What changes were proposed in this pull request?
This PR removes the ANTLR3 based parser, and moves the new ANTLR4 based parser into the `org.apache.spark.sql.catalyst.parser package`.
### How was this patch tested?
Existing unit tests.
cc rxin andrewor14 yhuai
Author: Herman van Hovell <hvanhovell@questtec.nl>
Closes #12071 from hvanhovell/SPARK-14211.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
Major changes:
1. Implement `FileFormat.buildReader()` for the CSV data source.
1. Add an extra argument to `FileFormat.buildReader()`, `physicalSchema`, which is basically the result of `FileFormat.inferSchema` or user specified schema.
This argument is necessary because the CSV data source needs to know all the columns of the underlying files to read the file.
## How was this patch tested?
Existing tests should do the work.
Author: Cheng Lian <lian@databricks.com>
Closes #12002 from liancheng/spark-14206-csv-build-reader.
|
| |
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This change resolves an issue where `DataFrameNaFunctions.fill` changes a `FloatType` column to a `DoubleType`. We also clarify the contract that replacement values will be cast to the column data type, which may change the replacement value when casting to a lower precision type.
## How was this patch tested?
This patch has associated unit tests.
Author: Travis Crawford <travis@medium.com>
Closes #11967 from traviscrawford/SPARK-14081-dataframena.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
properly
## What changes were proposed in this pull request?
This PR improves `CodeFormatter` to fix the following malformed indentations.
```java
/* 019 */ public java.lang.Object apply(java.lang.Object _i) {
/* 020 */ InternalRow i = (InternalRow) _i;
/* 021 */ /* createexternalrow(if (isnull(input[0, double])) null else input[0, double], if (isnull(input[1, int])) null else input[1, int], ... */
/* 022 */ boolean isNull = false;
/* 023 */ final Object[] values = new Object[2];
/* 024 */ /* if (isnull(input[0, double])) null else input[0, double] */
/* 025 */ /* isnull(input[0, double]) */
...
/* 053 */ if (!false && false) {
/* 054 */ /* null */
/* 055 */ final int value9 = -1;
/* 056 */ isNull6 = true;
/* 057 */ value6 = value9;
/* 058 */ } else {
...
/* 077 */ return mutableRow;
/* 078 */ }
/* 079 */ }
/* 080 */
```
After this PR, the code will be formatted like the following.
```java
/* 019 */ public java.lang.Object apply(java.lang.Object _i) {
/* 020 */ InternalRow i = (InternalRow) _i;
/* 021 */ /* createexternalrow(if (isnull(input[0, double])) null else input[0, double], if (isnull(input[1, int])) null else input[1, int], ... */
/* 022 */ boolean isNull = false;
/* 023 */ final Object[] values = new Object[2];
/* 024 */ /* if (isnull(input[0, double])) null else input[0, double] */
/* 025 */ /* isnull(input[0, double]) */
...
/* 053 */ if (!false && false) {
/* 054 */ /* null */
/* 055 */ final int value9 = -1;
/* 056 */ isNull6 = true;
/* 057 */ value6 = value9;
/* 058 */ } else {
...
/* 077 */ return mutableRow;
/* 078 */ }
/* 079 */ }
/* 080 */
```
Also, this issue fixes the following too. (Similar with [SPARK-14185](https://issues.apache.org/jira/browse/SPARK-14185))
```java
16/03/30 12:39:24 DEBUG WholeStageCodegen: /* 001 */ public Object generate(Object[] references) {
/* 002 */ return new GeneratedIterator(references);
/* 003 */ }
```
```java
16/03/30 12:46:32 DEBUG WholeStageCodegen:
/* 001 */ public Object generate(Object[] references) {
/* 002 */ return new GeneratedIterator(references);
/* 003 */ }
```
## How was this patch tested?
Pass the Jenkins tests (including new CodeFormatterSuite testcases.)
Author: Dongjoon Hyun <dongjoon@apache.org>
Closes #12072 from dongjoon-hyun/SPARK-14282.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
partition
## What changes were proposed in this pull request?
This pr is to add a config to control the maximum number of files as even small files have a non-trivial fixed cost. The current packing can put a lot of small files together which cases straggler tasks.
## How was this patch tested?
I added tests to check if many files get split into partitions in FileSourceStrategySuite.
Author: Takeshi YAMAMURO <linguin.m.s@gmail.com>
Closes #12068 from maropu/SPARK-14259.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
serializer and deserializer in ExpressionEncoder
## What changes were proposed in this pull request?
In `ExpressionEncoder`, we use `constructorFor` to build `fromRowExpression` as the `deserializer` in `ObjectOperator`. It's kind of confusing, we should make the name consistent.
## How was this patch tested?
existing tests.
Author: Wenchen Fan <wenchen@databricks.com>
Closes #12058 from cloud-fan/rename.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR implements buildReader for text data source and enable it in the new data source code path.
## How was this patch tested?
Existing tests.
Author: Wenchen Fan <wenchen@databricks.com>
Closes #11934 from cloud-fan/text.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
#### What changes were proposed in this pull request?
This PR is to implement the following four Database-related DDL commands:
- `CREATE DATABASE|SCHEMA [IF NOT EXISTS] database_name`
- `DROP DATABASE [IF EXISTS] database_name [RESTRICT|CASCADE]`
- `DESCRIBE DATABASE [EXTENDED] db_name`
- `ALTER (DATABASE|SCHEMA) database_name SET DBPROPERTIES (property_name=property_value, ...)`
Another PR will be submitted to handle the unsupported commands. In the Database-related DDL commands, we will issue an error exception for `ALTER (DATABASE|SCHEMA) database_name SET OWNER [USER|ROLE] user_or_role`.
cc yhuai andrewor14 rxin Could you review the changes? Is it in the right direction? Thanks!
#### How was this patch tested?
Added a few test cases in `command/DDLSuite.scala` for testing DDL command execution in `SQLContext`. Since `HiveContext` also shares the same implementation, the existing test cases in `\hive` also verifies the correctness of these commands.
Author: gatorsmile <gatorsmile@gmail.com>
Author: xiaoli <lixiao1983@gmail.com>
Author: Xiao Li <xiaoli@Xiaos-MacBook-Pro.local>
Closes #12009 from gatorsmile/dbDDL.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
Builds on https://github.com/apache/spark/pull/12022 and (a) appends "..." to truncated comment strings and (b) fixes indentation in lines after the commented strings if they happen to have a `(`, `{`, `)` or `}`
## How was this patch tested?
Manually examined the generated code.
Author: Sameer Agarwal <sameer@databricks.com>
Closes #12044 from sameeragarwal/comment.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR brings the support for chained Python UDFs, for example
```sql
select udf1(udf2(a))
select udf1(udf2(a) + 3)
select udf1(udf2(a) + udf3(b))
```
Also directly chained unary Python UDFs are put in single batch of Python UDFs, others may require multiple batches.
For example,
```python
>>> sqlContext.sql("select double(double(1))").explain()
== Physical Plan ==
WholeStageCodegen
: +- Project [pythonUDF#10 AS double(double(1))#9]
: +- INPUT
+- !BatchPythonEvaluation double(double(1)), [pythonUDF#10]
+- Scan OneRowRelation[]
>>> sqlContext.sql("select double(double(1) + double(2))").explain()
== Physical Plan ==
WholeStageCodegen
: +- Project [pythonUDF#19 AS double((double(1) + double(2)))#16]
: +- INPUT
+- !BatchPythonEvaluation double((pythonUDF#17 + pythonUDF#18)), [pythonUDF#17,pythonUDF#18,pythonUDF#19]
+- !BatchPythonEvaluation double(2), [pythonUDF#17,pythonUDF#18]
+- !BatchPythonEvaluation double(1), [pythonUDF#17]
+- Scan OneRowRelation[]
```
TODO: will support multiple unrelated Python UDFs in one batch (another PR).
## How was this patch tested?
Added new unit tests for chained UDFs.
Author: Davies Liu <davies@databricks.com>
Closes #12014 from davies/py_udfs.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This adds `debugCodegen` to the debug package for query execution.
## How was this patch tested?
Unit and manual testing. Output example:
```
scala> import org.apache.spark.sql.execution.debug._
import org.apache.spark.sql.execution.debug._
scala> sqlContext.range(100).groupBy("id").count().orderBy("id").debugCodegen()
Found 3 WholeStageCodegen subtrees.
== Subtree 1 / 3 ==
WholeStageCodegen
: +- TungstenAggregate(key=[id#0L], functions=[(count(1),mode=Partial,isDistinct=false)], output=[id#0L,count#9L])
: +- Range 0, 1, 1, 100, [id#0L]
Generated code:
/* 001 */ public Object generate(Object[] references) {
/* 002 */ return new GeneratedIterator(references);
/* 003 */ }
/* 004 */
/* 005 */ /** Codegened pipeline for:
/* 006 */ * TungstenAggregate(key=[id#0L], functions=[(count(1),mode=Partial,isDistinct=false)], output=[id#0L,count#9L])
/* 007 */ +- Range 0, 1, 1, 100, [id#0L]
/* 008 */ */
/* 009 */ final class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {
/* 010 */ private Object[] references;
/* 011 */ private boolean agg_initAgg;
/* 012 */ private org.apache.spark.sql.execution.aggregate.TungstenAggregate agg_plan;
/* 013 */ private org.apache.spark.sql.execution.UnsafeFixedWidthAggregationMap agg_hashMap;
/* 014 */ private org.apache.spark.sql.execution.UnsafeKVExternalSorter agg_sorter;
/* 015 */ private org.apache.spark.unsafe.KVIterator agg_mapIter;
/* 016 */ private org.apache.spark.sql.execution.metric.LongSQLMetric range_numOutputRows;
/* 017 */ private org.apache.spark.sql.execution.metric.LongSQLMetricValue range_metricValue;
/* 018 */ private boolean range_initRange;
/* 019 */ private long range_partitionEnd;
/* 020 */ private long range_number;
/* 021 */ private boolean range_overflow;
/* 022 */ private scala.collection.Iterator range_input;
/* 023 */ private UnsafeRow range_result;
/* 024 */ private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder range_holder;
/* 025 */ private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter range_rowWriter;
/* 026 */ private UnsafeRow agg_result;
/* 027 */ private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder agg_holder;
/* 028 */ private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter agg_rowWriter;
/* 029 */ private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowJoiner agg_unsafeRowJoiner;
/* 030 */ private org.apache.spark.sql.execution.metric.LongSQLMetric wholestagecodegen_numOutputRows;
/* 031 */ private org.apache.spark.sql.execution.metric.LongSQLMetricValue wholestagecodegen_metricValue;
/* 032 */
/* 033 */ public GeneratedIterator(Object[] references) {
/* 034 */ this.references = references;
/* 035 */ }
/* 036 */
/* 037 */ public void init(scala.collection.Iterator inputs[]) {
/* 038 */ agg_initAgg = false;
/* 039 */ this.agg_plan = (org.apache.spark.sql.execution.aggregate.TungstenAggregate) references[0];
/* 040 */ agg_hashMap = agg_plan.createHashMap();
/* 041 */
/* 042 */ this.range_numOutputRows = (org.apache.spark.sql.execution.metric.LongSQLMetric) references[1];
/* 043 */ range_metricValue = (org.apache.spark.sql.execution.metric.LongSQLMetricValue) range_numOutputRows.localValue();
/* 044 */ range_initRange = false;
/* 045 */ range_partitionEnd = 0L;
/* 046 */ range_number = 0L;
/* 047 */ range_overflow = false;
/* 048 */ range_input = inputs[0];
/* 049 */ range_result = new UnsafeRow(1);
/* 050 */ this.range_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(range_result, 0);
/* 051 */ this.range_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(range_holder, 1);
/* 052 */ agg_result = new UnsafeRow(1);
/* 053 */ this.agg_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(agg_result, 0);
/* 054 */ this.agg_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(agg_holder, 1);
/* 055 */ agg_unsafeRowJoiner = agg_plan.createUnsafeJoiner();
/* 056 */ this.wholestagecodegen_numOutputRows = (org.apache.spark.sql.execution.metric.LongSQLMetric) references[2];
/* 057 */ wholestagecodegen_metricValue = (org.apache.spark.sql.execution.metric.LongSQLMetricValue) wholestagecodegen_numOutputRows.localValue();
/* 058 */ }
/* 059 */
/* 060 */ private void agg_doAggregateWithKeys() throws java.io.IOException {
/* 061 */ /*** PRODUCE: Range 0, 1, 1, 100, [id#0L] */
/* 062 */
/* 063 */ // initialize Range
/* 064 */ if (!range_initRange) {
/* 065 */ range_initRange = true;
/* 066 */ if (range_input.hasNext()) {
/* 067 */ initRange(((InternalRow) range_input.next()).getInt(0));
/* 068 */ } else {
/* 069 */ return;
/* 070 */ }
/* 071 */ }
/* 072 */
/* 073 */ while (!range_overflow && range_number < range_partitionEnd) {
/* 074 */ long range_value = range_number;
/* 075 */ range_number += 1L;
/* 076 */ if (range_number < range_value ^ 1L < 0) {
/* 077 */ range_overflow = true;
/* 078 */ }
/* 079 */
/* 080 */ /*** CONSUME: TungstenAggregate(key=[id#0L], functions=[(count(1),mode=Partial,isDistinct=false)], output=[id#0L,count#9L]) */
/* 081 */
/* 082 */ // generate grouping key
/* 083 */ agg_rowWriter.write(0, range_value);
/* 084 */ /* hash(input[0, bigint], 42) */
/* 085 */ int agg_value1 = 42;
/* 086 */
/* 087 */ agg_value1 = org.apache.spark.unsafe.hash.Murmur3_x86_32.hashLong(range_value, agg_value1);
/* 088 */ UnsafeRow agg_aggBuffer = null;
/* 089 */ if (true) {
/* 090 */ // try to get the buffer from hash map
/* 091 */ agg_aggBuffer = agg_hashMap.getAggregationBufferFromUnsafeRow(agg_result, agg_value1);
/* 092 */ }
/* 093 */ if (agg_aggBuffer == null) {
/* 094 */ if (agg_sorter == null) {
/* 095 */ agg_sorter = agg_hashMap.destructAndCreateExternalSorter();
/* 096 */ } else {
/* 097 */ agg_sorter.merge(agg_hashMap.destructAndCreateExternalSorter());
/* 098 */ }
/* 099 */
/* 100 */ // the hash map had be spilled, it should have enough memory now,
/* 101 */ // try to allocate buffer again.
/* 102 */ agg_aggBuffer = agg_hashMap.getAggregationBufferFromUnsafeRow(agg_result, agg_value1);
/* 103 */ if (agg_aggBuffer == null) {
/* 104 */ // failed to allocate the first page
/* 105 */ throw new OutOfMemoryError("No enough memory for aggregation");
/* 106 */ }
/* 107 */ }
/* 108 */
/* 109 */ // evaluate aggregate function
/* 110 */ /* (input[0, bigint] + 1) */
/* 111 */ /* input[0, bigint] */
/* 112 */ long agg_value4 = agg_aggBuffer.getLong(0);
/* 113 */
/* 114 */ long agg_value3 = -1L;
/* 115 */ agg_value3 = agg_value4 + 1L;
/* 116 */ // update aggregate buffer
/* 117 */ agg_aggBuffer.setLong(0, agg_value3);
/* 118 */
/* 119 */ if (shouldStop()) return;
/* 120 */ }
/* 121 */
/* 122 */ agg_mapIter = agg_plan.finishAggregate(agg_hashMap, agg_sorter);
/* 123 */ }
/* 124 */
/* 125 */ private void initRange(int idx) {
/* 126 */ java.math.BigInteger index = java.math.BigInteger.valueOf(idx);
/* 127 */ java.math.BigInteger numSlice = java.math.BigInteger.valueOf(1L);
/* 128 */ java.math.BigInteger numElement = java.math.BigInteger.valueOf(100L);
/* 129 */ java.math.BigInteger step = java.math.BigInteger.valueOf(1L);
/* 130 */ java.math.BigInteger start = java.math.BigInteger.valueOf(0L);
/* 131 */
/* 132 */ java.math.BigInteger st = index.multiply(numElement).divide(numSlice).multiply(step).add(start);
/* 133 */ if (st.compareTo(java.math.BigInteger.valueOf(Long.MAX_VALUE)) > 0) {
/* 134 */ range_number = Long.MAX_VALUE;
/* 135 */ } else if (st.compareTo(java.math.BigInteger.valueOf(Long.MIN_VALUE)) < 0) {
/* 136 */ range_number = Long.MIN_VALUE;
/* 137 */ } else {
/* 138 */ range_number = st.longValue();
/* 139 */ }
/* 140 */
/* 141 */ java.math.BigInteger end = index.add(java.math.BigInteger.ONE).multiply(numElement).divide(numSlice)
/* 142 */ .multiply(step).add(start);
/* 143 */ if (end.compareTo(java.math.BigInteger.valueOf(Long.MAX_VALUE)) > 0) {
/* 144 */ range_partitionEnd = Long.MAX_VALUE;
/* 145 */ } else if (end.compareTo(java.math.BigInteger.valueOf(Long.MIN_VALUE)) < 0) {
/* 146 */ range_partitionEnd = Long.MIN_VALUE;
/* 147 */ } else {
/* 148 */ range_partitionEnd = end.longValue();
/* 149 */ }
/* 150 */
/* 151 */ range_metricValue.add((range_partitionEnd - range_number) / 1L);
/* 152 */ }
/* 153 */
/* 154 */ protected void processNext() throws java.io.IOException {
/* 155 */ /*** PRODUCE: TungstenAggregate(key=[id#0L], functions=[(count(1),mode=Partial,isDistinct=false)], output=[id#0L,count#9L]) */
/* 156 */
/* 157 */ if (!agg_initAgg) {
/* 158 */ agg_initAgg = true;
/* 159 */ agg_doAggregateWithKeys();
/* 160 */ }
/* 161 */
/* 162 */ // output the result
/* 163 */ while (agg_mapIter.next()) {
/* 164 */ wholestagecodegen_metricValue.add(1);
/* 165 */ UnsafeRow agg_aggKey = (UnsafeRow) agg_mapIter.getKey();
/* 166 */ UnsafeRow agg_aggBuffer1 = (UnsafeRow) agg_mapIter.getValue();
/* 167 */
/* 168 */ UnsafeRow agg_resultRow = agg_unsafeRowJoiner.join(agg_aggKey, agg_aggBuffer1);
/* 169 */
/* 170 */ /*** CONSUME: WholeStageCodegen */
/* 171 */
/* 172 */ append(agg_resultRow);
/* 173 */
/* 174 */ if (shouldStop()) return;
/* 175 */ }
/* 176 */
/* 177 */ agg_mapIter.close();
/* 178 */ if (agg_sorter == null) {
/* 179 */ agg_hashMap.free();
/* 180 */ }
/* 181 */ }
/* 182 */ }
== Subtree 2 / 3 ==
WholeStageCodegen
: +- Sort [id#0L ASC], true, 0
: +- INPUT
+- Exchange rangepartitioning(id#0L ASC, 200), None
+- WholeStageCodegen
: +- TungstenAggregate(key=[id#0L], functions=[(count(1),mode=Final,isDistinct=false)], output=[id#0L,count#4L])
: +- INPUT
+- Exchange hashpartitioning(id#0L, 200), None
+- WholeStageCodegen
: +- TungstenAggregate(key=[id#0L], functions=[(count(1),mode=Partial,isDistinct=false)], output=[id#0L,count#9L])
: +- Range 0, 1, 1, 100, [id#0L]
Generated code:
/* 001 */ public Object generate(Object[] references) {
/* 002 */ return new GeneratedIterator(references);
/* 003 */ }
/* 004 */
/* 005 */ /** Codegened pipeline for:
/* 006 */ * Sort [id#0L ASC], true, 0
/* 007 */ +- INPUT
/* 008 */ */
/* 009 */ final class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {
/* 010 */ private Object[] references;
/* 011 */ private boolean sort_needToSort;
/* 012 */ private org.apache.spark.sql.execution.Sort sort_plan;
/* 013 */ private org.apache.spark.sql.execution.UnsafeExternalRowSorter sort_sorter;
/* 014 */ private org.apache.spark.executor.TaskMetrics sort_metrics;
/* 015 */ private scala.collection.Iterator<UnsafeRow> sort_sortedIter;
/* 016 */ private scala.collection.Iterator inputadapter_input;
/* 017 */ private org.apache.spark.sql.execution.metric.LongSQLMetric sort_dataSize;
/* 018 */ private org.apache.spark.sql.execution.metric.LongSQLMetricValue sort_metricValue;
/* 019 */ private org.apache.spark.sql.execution.metric.LongSQLMetric sort_spillSize;
/* 020 */ private org.apache.spark.sql.execution.metric.LongSQLMetricValue sort_metricValue1;
/* 021 */
/* 022 */ public GeneratedIterator(Object[] references) {
/* 023 */ this.references = references;
/* 024 */ }
/* 025 */
/* 026 */ public void init(scala.collection.Iterator inputs[]) {
/* 027 */ sort_needToSort = true;
/* 028 */ this.sort_plan = (org.apache.spark.sql.execution.Sort) references[0];
/* 029 */ sort_sorter = sort_plan.createSorter();
/* 030 */ sort_metrics = org.apache.spark.TaskContext.get().taskMetrics();
/* 031 */
/* 032 */ inputadapter_input = inputs[0];
/* 033 */ this.sort_dataSize = (org.apache.spark.sql.execution.metric.LongSQLMetric) references[1];
/* 034 */ sort_metricValue = (org.apache.spark.sql.execution.metric.LongSQLMetricValue) sort_dataSize.localValue();
/* 035 */ this.sort_spillSize = (org.apache.spark.sql.execution.metric.LongSQLMetric) references[2];
/* 036 */ sort_metricValue1 = (org.apache.spark.sql.execution.metric.LongSQLMetricValue) sort_spillSize.localValue();
/* 037 */ }
/* 038 */
/* 039 */ private void sort_addToSorter() throws java.io.IOException {
/* 040 */ /*** PRODUCE: INPUT */
/* 041 */
/* 042 */ while (inputadapter_input.hasNext()) {
/* 043 */ InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();
/* 044 */ /*** CONSUME: Sort [id#0L ASC], true, 0 */
/* 045 */
/* 046 */ sort_sorter.insertRow((UnsafeRow)inputadapter_row);
/* 047 */ if (shouldStop()) return;
/* 048 */ }
/* 049 */
/* 050 */ }
/* 051 */
/* 052 */ protected void processNext() throws java.io.IOException {
/* 053 */ /*** PRODUCE: Sort [id#0L ASC], true, 0 */
/* 054 */ if (sort_needToSort) {
/* 055 */ sort_addToSorter();
/* 056 */ Long sort_spillSizeBefore = sort_metrics.memoryBytesSpilled();
/* 057 */ sort_sortedIter = sort_sorter.sort();
/* 058 */ sort_metricValue.add(sort_sorter.getPeakMemoryUsage());
/* 059 */ sort_metricValue1.add(sort_metrics.memoryBytesSpilled() - sort_spillSizeBefore);
/* 060 */ sort_metrics.incPeakExecutionMemory(sort_sorter.getPeakMemoryUsage());
/* 061 */ sort_needToSort = false;
/* 062 */ }
/* 063 */
/* 064 */ while (sort_sortedIter.hasNext()) {
/* 065 */ UnsafeRow sort_outputRow = (UnsafeRow)sort_sortedIter.next();
/* 066 */
/* 067 */ /*** CONSUME: WholeStageCodegen */
/* 068 */
/* 069 */ append(sort_outputRow);
/* 070 */
/* 071 */ if (shouldStop()) return;
/* 072 */ }
/* 073 */ }
/* 074 */ }
== Subtree 3 / 3 ==
WholeStageCodegen
: +- TungstenAggregate(key=[id#0L], functions=[(count(1),mode=Final,isDistinct=false)], output=[id#0L,count#4L])
: +- INPUT
+- Exchange hashpartitioning(id#0L, 200), None
+- WholeStageCodegen
: +- TungstenAggregate(key=[id#0L], functions=[(count(1),mode=Partial,isDistinct=false)], output=[id#0L,count#9L])
: +- Range 0, 1, 1, 100, [id#0L]
Generated code:
/* 001 */ public Object generate(Object[] references) {
/* 002 */ return new GeneratedIterator(references);
/* 003 */ }
/* 004 */
/* 005 */ /** Codegened pipeline for:
/* 006 */ * TungstenAggregate(key=[id#0L], functions=[(count(1),mode=Final,isDistinct=false)], output=[id#0L,count#4L])
/* 007 */ +- INPUT
/* 008 */ */
/* 009 */ final class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {
/* 010 */ private Object[] references;
/* 011 */ private boolean agg_initAgg;
/* 012 */ private org.apache.spark.sql.execution.aggregate.TungstenAggregate agg_plan;
/* 013 */ private org.apache.spark.sql.execution.UnsafeFixedWidthAggregationMap agg_hashMap;
/* 014 */ private org.apache.spark.sql.execution.UnsafeKVExternalSorter agg_sorter;
/* 015 */ private org.apache.spark.unsafe.KVIterator agg_mapIter;
/* 016 */ private scala.collection.Iterator inputadapter_input;
/* 017 */ private UnsafeRow agg_result;
/* 018 */ private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder agg_holder;
/* 019 */ private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter agg_rowWriter;
/* 020 */ private UnsafeRow agg_result1;
/* 021 */ private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder agg_holder1;
/* 022 */ private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter agg_rowWriter1;
/* 023 */ private org.apache.spark.sql.execution.metric.LongSQLMetric wholestagecodegen_numOutputRows;
/* 024 */ private org.apache.spark.sql.execution.metric.LongSQLMetricValue wholestagecodegen_metricValue;
/* 025 */
/* 026 */ public GeneratedIterator(Object[] references) {
/* 027 */ this.references = references;
/* 028 */ }
/* 029 */
/* 030 */ public void init(scala.collection.Iterator inputs[]) {
/* 031 */ agg_initAgg = false;
/* 032 */ this.agg_plan = (org.apache.spark.sql.execution.aggregate.TungstenAggregate) references[0];
/* 033 */ agg_hashMap = agg_plan.createHashMap();
/* 034 */
/* 035 */ inputadapter_input = inputs[0];
/* 036 */ agg_result = new UnsafeRow(1);
/* 037 */ this.agg_holder = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(agg_result, 0);
/* 038 */ this.agg_rowWriter = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(agg_holder, 1);
/* 039 */ agg_result1 = new UnsafeRow(2);
/* 040 */ this.agg_holder1 = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(agg_result1, 0);
/* 041 */ this.agg_rowWriter1 = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(agg_holder1, 2);
/* 042 */ this.wholestagecodegen_numOutputRows = (org.apache.spark.sql.execution.metric.LongSQLMetric) references[1];
/* 043 */ wholestagecodegen_metricValue = (org.apache.spark.sql.execution.metric.LongSQLMetricValue) wholestagecodegen_numOutputRows.localValue();
/* 044 */ }
/* 045 */
/* 046 */ private void agg_doAggregateWithKeys() throws java.io.IOException {
/* 047 */ /*** PRODUCE: INPUT */
/* 048 */
/* 049 */ while (inputadapter_input.hasNext()) {
/* 050 */ InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();
/* 051 */ /*** CONSUME: TungstenAggregate(key=[id#0L], functions=[(count(1),mode=Final,isDistinct=false)], output=[id#0L,count#4L]) */
/* 052 */ /* input[0, bigint] */
/* 053 */ long inputadapter_value = inputadapter_row.getLong(0);
/* 054 */ /* input[1, bigint] */
/* 055 */ long inputadapter_value1 = inputadapter_row.getLong(1);
/* 056 */
/* 057 */ // generate grouping key
/* 058 */ agg_rowWriter.write(0, inputadapter_value);
/* 059 */ /* hash(input[0, bigint], 42) */
/* 060 */ int agg_value1 = 42;
/* 061 */
/* 062 */ agg_value1 = org.apache.spark.unsafe.hash.Murmur3_x86_32.hashLong(inputadapter_value, agg_value1);
/* 063 */ UnsafeRow agg_aggBuffer = null;
/* 064 */ if (true) {
/* 065 */ // try to get the buffer from hash map
/* 066 */ agg_aggBuffer = agg_hashMap.getAggregationBufferFromUnsafeRow(agg_result, agg_value1);
/* 067 */ }
/* 068 */ if (agg_aggBuffer == null) {
/* 069 */ if (agg_sorter == null) {
/* 070 */ agg_sorter = agg_hashMap.destructAndCreateExternalSorter();
/* 071 */ } else {
/* 072 */ agg_sorter.merge(agg_hashMap.destructAndCreateExternalSorter());
/* 073 */ }
/* 074 */
/* 075 */ // the hash map had be spilled, it should have enough memory now,
/* 076 */ // try to allocate buffer again.
/* 077 */ agg_aggBuffer = agg_hashMap.getAggregationBufferFromUnsafeRow(agg_result, agg_value1);
/* 078 */ if (agg_aggBuffer == null) {
/* 079 */ // failed to allocate the first page
/* 080 */ throw new OutOfMemoryError("No enough memory for aggregation");
/* 081 */ }
/* 082 */ }
/* 083 */
/* 084 */ // evaluate aggregate function
/* 085 */ /* (input[0, bigint] + input[2, bigint]) */
/* 086 */ /* input[0, bigint] */
/* 087 */ long agg_value4 = agg_aggBuffer.getLong(0);
/* 088 */
/* 089 */ long agg_value3 = -1L;
/* 090 */ agg_value3 = agg_value4 + inputadapter_value1;
/* 091 */ // update aggregate buffer
/* 092 */ agg_aggBuffer.setLong(0, agg_value3);
/* 093 */ if (shouldStop()) return;
/* 094 */ }
/* 095 */
/* 096 */ agg_mapIter = agg_plan.finishAggregate(agg_hashMap, agg_sorter);
/* 097 */ }
/* 098 */
/* 099 */ protected void processNext() throws java.io.IOException {
/* 100 */ /*** PRODUCE: TungstenAggregate(key=[id#0L], functions=[(count(1),mode=Final,isDistinct=false)], output=[id#0L,count#4L]) */
/* 101 */
/* 102 */ if (!agg_initAgg) {
/* 103 */ agg_initAgg = true;
/* 104 */ agg_doAggregateWithKeys();
/* 105 */ }
/* 106 */
/* 107 */ // output the result
/* 108 */ while (agg_mapIter.next()) {
/* 109 */ wholestagecodegen_metricValue.add(1);
/* 110 */ UnsafeRow agg_aggKey = (UnsafeRow) agg_mapIter.getKey();
/* 111 */ UnsafeRow agg_aggBuffer1 = (UnsafeRow) agg_mapIter.getValue();
/* 112 */
/* 113 */ /* input[0, bigint] */
/* 114 */ long agg_value6 = agg_aggKey.getLong(0);
/* 115 */ /* input[0, bigint] */
/* 116 */ long agg_value7 = agg_aggBuffer1.getLong(0);
/* 117 */
/* 118 */ /*** CONSUME: WholeStageCodegen */
/* 119 */
/* 120 */ agg_rowWriter1.write(0, agg_value6);
/* 121 */
/* 122 */ agg_rowWriter1.write(1, agg_value7);
/* 123 */ append(agg_result1);
/* 124 */
/* 125 */ if (shouldStop()) return;
/* 126 */ }
/* 127 */
/* 128 */ agg_mapIter.close();
/* 129 */ if (agg_sorter == null) {
/* 130 */ agg_hashMap.free();
/* 131 */ }
/* 132 */ }
/* 133 */ }
```
rxin
Author: Eric Liang <ekl@databricks.com>
Closes #12025 from ericl/spark-14227.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR is a simple fix for an exception message to print `string[]` content correctly.
```java
String[] colPath = requestedSchema.getPaths().get(i);
...
- throw new IOException("Required column is missing in data file. Col: " + colPath);
+ throw new IOException("Required column is missing in data file. Col: " + Arrays.toString(colPath));
```
## How was this patch tested?
Manual.
Author: Dongjoon Hyun <dongjoon@apache.org>
Closes #12041 from dongjoon-hyun/fix_exception_message_with_string_array.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR fixes two trivial typos: 'does not **much**' --> 'does not **match**'.
## How was this patch tested?
Manual.
Author: Dongjoon Hyun <dongjoon@apache.org>
Closes #12042 from dongjoon-hyun/fix_typo_by_replacing_much_with_match.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
Renames SQL option `spark.sql.parquet.fileScan` since now all `HadoopFsRelation` based data sources are being migrated to `FileScanRDD` code path.
## How was this patch tested?
None.
Author: Cheng Lian <lian@databricks.com>
Closes #12003 from liancheng/spark-14208-option-renaming.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR implements buildReader for json data source and enable it in the new data source code path.
## How was this patch tested?
existing tests
Author: Wenchen Fan <wenchen@databricks.com>
Closes #11960 from cloud-fan/json.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This adds a metric to parquet scans that measures the time in just the scan phase. This is
only possible when the scan returns ColumnarBatches, otherwise the overhead is too high.
This combined with the pipeline metric lets us easily see what percent of the time was
in the scan.
Author: Nong Li <nong@databricks.com>
Closes #12007 from nongli/spark-14210.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This improves the Filter codegen for NULLs by deferring loading the values for IsNotNull.
Instead of generating code like:
boolean isNull = ...
int value = ...
if (isNull) continue;
we will generate:
boolean isNull = ...
if (isNull) continue;
int value = ...
This is useful since retrieving the values can be non-trivial (they can be dictionary encoded
among other things). This currently only works when the attribute comes from the column batch
but could be extended to other cases in the future.
## How was this patch tested?
On tpcds q55, this fixes the regression from introducing the IsNotNull predicates.
```
TPCDS Snappy: Best/Avg Time(ms) Rate(M/s) Per Row(ns)
--------------------------------------------------------------------------------
q55 4564 / 5036 25.2 39.6
q55 4064 / 4340 28.3 35.3
```
Author: Nong Li <nong@databricks.com>
Closes #11792 from nongli/spark-13981.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
### What changes were proposed in this pull request?
This PR migrates all HiveQl parsing to the new ANTLR4 parser. This PR is build on top of https://github.com/apache/spark/pull/12011, and we should wait with merging until that one is in (hence the WIP tag).
As soon as this PR is merged we can start removing much of the old parser infrastructure.
### How was this patch tested?
Exisiting Hive unit tests.
cc rxin andrewor14 yhuai
Author: Herman van Hovell <hvanhovell@questtec.nl>
Closes #12015 from hvanhovell/SPARK-14213.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
After DataFrame and Dataset are merged, the trait `Queryable` becomes unnecessary as it has only one implementation. We should remove it.
## How was this patch tested?
existing tests.
Author: Wenchen Fan <wenchen@databricks.com>
Closes #12001 from cloud-fan/df-ds.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
DDL commands
## What changes were proposed in this pull request?
Before: We just pass all role commands to Hive even though it doesn't work.
After: We throw an `AnalysisException` that looks like this:
```
scala> sql("CREATE ROLE x")
org.apache.spark.sql.AnalysisException: Unsupported Hive operation: CREATE ROLE;
at org.apache.spark.sql.hive.HiveQl$$anonfun$parsePlan$1.apply(HiveQl.scala:213)
at org.apache.spark.sql.hive.HiveQl$$anonfun$parsePlan$1.apply(HiveQl.scala:208)
at org.apache.spark.sql.catalyst.parser.CatalystQl.safeParse(CatalystQl.scala:49)
at org.apache.spark.sql.hive.HiveQl.parsePlan(HiveQl.scala:208)
at org.apache.spark.sql.SQLContext.parseSql(SQLContext.scala:198)
```
## How was this patch tested?
`HiveQuerySuite`
Author: Andrew Or <andrew@databricks.com>
Closes #11948 from andrewor14/ddl-role-management.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
Session catalog was added in #11750. However, it doesn't really support temporary functions properly; right now we only store the metadata in the form of `CatalogFunction`, but this doesn't make sense for temporary functions because there is no class name.
This patch moves the `FunctionRegistry` into the `SessionCatalog`. With this, the user can call `catalog.createTempFunction` and `catalog.lookupFunction` to use the function they registered previously. This is currently still dead code, however.
## How was this patch tested?
`SessionCatalogSuite`.
Author: Andrew Or <andrew@databricks.com>
Closes #11972 from andrewor14/temp-functions.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
Extract the workaround for HADOOP-10622 introduced by #11940 into UninterruptibleThread so that we can test and reuse it.
## How was this patch tested?
Unit tests
Author: Shixiong Zhu <shixiong@databricks.com>
Closes #11971 from zsxwing/uninterrupt.
|
| |
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
UserDefinedType is a developer API in Spark 1.x. With very high probability we will create a new API for user-defined type that also works well with column batches as well as encoders (datasets). In Spark 2.0, let's make `UserDefinedType` `private[spark]` first.
## How was this patch tested?
Existing unit tests.
Author: Reynold Xin <rxin@databricks.com>
Closes #11955 from rxin/SPARK-14155.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This patch addresses the remaining comments left in #11750 and #11918 after they are merged. For a full list of changes in this patch, just trace the commits.
## How was this patch tested?
`SessionCatalogSuite` and `CatalogTestCases`
Author: Andrew Or <andrew@databricks.com>
Closes #12006 from andrewor14/session-catalog-followup.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
#### What changes were proposed in this pull request?
This PR adds all the current Spark SQL DDL commands to the new ANTLR 4 based SQL parser.
I have found a few inconsistencies in the current commands:
- Function has an alias field. This is actually the class name of the function.
- Partition specifications should contain nulls in some commands, and contain `None`s in others.
- `AlterTableSkewedLocation`: Should defines which columns have skewed values, and should allow us to define storage for each skewed combination of values. We currently only allow one value per field.
- `AlterTableSetFileFormat`: Should only have one file format, it currently supports both.
I have implemented all these comments like they were, and I propose to improve them in follow-up PRs.
#### How was this patch tested?
The existing DDLCommandSuite.
cc rxin andrewor14 yhuai
Author: Herman van Hovell <hvanhovell@questtec.nl>
Closes #12011 from hvanhovell/SPARK-14086.
|
| |
|
|
|
|
|
|
| |
Seems https://github.com/apache/spark/commit/600c0b69cab4767e8e5a6f4284777d8b9d4bd40e is missing the antlr4 maven plugin. This pr adds it.
Author: Yin Huai <yhuai@databricks.com>
Closes #12010 from yhuai/mavenAntlr4.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
Currently, for the key that can not fit within a long, we build a hash map for UnsafeHashedRelation, it's converted to BytesToBytesMap after serialization and deserialization. We should build a BytesToBytesMap directly to have better memory efficiency.
In order to do that, BytesToBytesMap should support multiple (K,V) pair with the same K, Location.putNewKey() is renamed to Location.append(), which could append multiple values for the same key (same Location). `Location.newValue()` is added to find the next value for the same key.
## How was this patch tested?
Existing tests. Added benchmark for broadcast hash join with duplicated keys.
Author: Davies Liu <davies@databricks.com>
Closes #11870 from davies/map2.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
### What changes were proposed in this pull request?
The current ANTLR3 parser is quite complex to maintain and suffers from code blow-ups. This PR introduces a new parser that is based on ANTLR4.
This parser is based on the [Presto's SQL parser](https://github.com/facebook/presto/blob/master/presto-parser/src/main/antlr4/com/facebook/presto/sql/parser/SqlBase.g4). The current implementation can parse and create Catalyst and SQL plans. Large parts of the HiveQl DDL and some of the DML functionality is currently missing, the plan is to add this in follow-up PRs.
This PR is a work in progress, and work needs to be done in the following area's:
- [x] Error handling should be improved.
- [x] Documentation should be improved.
- [x] Multi-Insert needs to be tested.
- [ ] Naming and package locations.
### How was this patch tested?
Catalyst and SQL unit tests.
Author: Herman van Hovell <hvanhovell@questtec.nl>
Closes #11557 from hvanhovell/ngParser.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
of computed tables
## What changes were proposed in this pull request?
JIRA: https://issues.apache.org/jira/browse/SPARK-14156
In HiveComparisonTest, when catalyst results are different to hive results, we will collect the messages for computed tables during the test. During creating the message, we use sparkPlan. But we actually run the query with executedPlan. So the error message is sometimes confusing.
For example, as wholestage codegen is enabled by default now. The shown spark plan for computed tables is the plan before wholestage codegen.
A concrete is the following error message shown before this patch. It is the error shown when running `HiveCompatibilityTest` `auto_join26`.
auto_join26 has one SQL to create table:
INSERT OVERWRITE TABLE dest_j1
SELECT x.key, count(1) FROM src1 x JOIN src y ON (x.key = y.key) group by x.key; (1)
Then a SQL to retrieve the result:
select * from dest_j1 x order by x.key; (2)
When the above SQL (2) to retrieve the result fails, In `HiveComparisonTest` we will try to collect and show the generated data from table `dest_j1` using the SQL (1)'s spark plan. The you will see this error:
TungstenAggregate(key=[key#8804], functions=[(count(1),mode=Partial,isDistinct=false)], output=[key#8804,count#8834L])
+- Project [key#8804]
+- BroadcastHashJoin [key#8804], [key#8806], Inner, BuildRight, None
:- Filter isnotnull(key#8804)
: +- InMemoryColumnarTableScan [key#8804], [isnotnull(key#8804)], InMemoryRelation [key#8804,value#8805], true, 5, StorageLevel(true, true, false, true, 1), HiveTableScan [key#8717,value#8718], MetastoreRelation default, src1, None, Some(src1)
+- Filter isnotnull(key#8806)
+- InMemoryColumnarTableScan [key#8806], [isnotnull(key#8806)], InMemoryRelation [key#8806,value#8807], true, 5, StorageLevel(true, true, false, true, 1), HiveTableScan [key#8760,value#8761], MetastoreRelation default, src, None, Some(src)
at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:47)
at org.apache.spark.sql.execution.aggregate.TungstenAggregate.doExecute(TungstenAggregate.scala:82)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:121)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:121)
at org.apache.spark.sql.execution.SparkPlan$$anonfun$executeQuery$1.apply(SparkPlan.scala:140)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.sql.execution.SparkPlan.executeQuery(SparkPlan.scala:137)
at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:120)
at org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1.apply(TungstenAggregate.scala:87)
at org.apache.spark.sql.execution.aggregate.TungstenAggregate$$anonfun$doExecute$1.apply(TungstenAggregate.scala:82)
at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:46)
... 70 more
Caused by: java.lang.UnsupportedOperationException: Filter does not implement doExecuteBroadcast
at org.apache.spark.sql.execution.SparkPlan.doExecuteBroadcast(SparkPlan.scala:221)
The message is confusing because it is not the plan actually run by SparkSQL engine to create the generated table. The plan actually run is no problem. But as before this patch, we run `e.sparkPlan.collect` to retrieve and show the generated data, spark plan is not the plan we can run. So the above error will be shown.
After this patch, we won't see the error because the executed plan is no problem and works.
## How was this patch tested?
Existing tests.
Author: Liang-Chi Hsieh <simonh@tw.ibm.com>
Closes #11957 from viirya/use-executedplan.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR simplifies generated code with a non-nullable column. This PR addresses three items:
1. Generate simplified code for and / or
2. Generate better code for divide and remainder with non-zero dividend
3. Pass nullable information into BoundReference at WholeStageCodegen
I have attached the generated code with and without this PR
## How was this patch tested?
Tested by existing test suites in sql/core
Here is a motivating example
````
(0 to 6).map(i => (i.toString, i.toInt)).toDF("k", "v")
.filter("v % 2 == 0").filter("v <= 4").filter("v > 1").show()
````
Generated code without this PR
````java
/* 032 */ protected void processNext() throws java.io.IOException {
/* 033 */ /*** PRODUCE: Project [_1#0 AS k#3,_2#1 AS v#4] */
/* 034 */
/* 035 */ /*** PRODUCE: Filter ((isnotnull((_2#1 % 2)) && ((_2#1 % 2) = 0)) && ((_2#1 <= 4) && (_2#1 > 1))) */
/* 036 */
/* 037 */ /*** PRODUCE: INPUT */
/* 038 */
/* 039 */ while (!shouldStop() && inputadapter_input.hasNext()) {
/* 040 */ InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();
/* 041 */ /*** CONSUME: Filter ((isnotnull((_2#1 % 2)) && ((_2#1 % 2) = 0)) && ((_2#1 <= 4) && (_2#1 > 1))) */
/* 042 */ /* input[1, int] */
/* 043 */ int filter_value1 = inputadapter_row.getInt(1);
/* 044 */
/* 045 */ /* isnotnull((input[1, int] % 2)) */
/* 046 */ /* (input[1, int] % 2) */
/* 047 */ boolean filter_isNull3 = false;
/* 048 */ int filter_value3 = -1;
/* 049 */ if (false || 2 == 0) {
/* 050 */ filter_isNull3 = true;
/* 051 */ } else {
/* 052 */ if (false) {
/* 053 */ filter_isNull3 = true;
/* 054 */ } else {
/* 055 */ filter_value3 = (int)(filter_value1 % 2);
/* 056 */ }
/* 057 */ }
/* 058 */ if (!(!(filter_isNull3))) continue;
/* 059 */
/* 060 */ /* ((input[1, int] % 2) = 0) */
/* 061 */ boolean filter_isNull6 = true;
/* 062 */ boolean filter_value6 = false;
/* 063 */ /* (input[1, int] % 2) */
/* 064 */ boolean filter_isNull7 = false;
/* 065 */ int filter_value7 = -1;
/* 066 */ if (false || 2 == 0) {
/* 067 */ filter_isNull7 = true;
/* 068 */ } else {
/* 069 */ if (false) {
/* 070 */ filter_isNull7 = true;
/* 071 */ } else {
/* 072 */ filter_value7 = (int)(filter_value1 % 2);
/* 073 */ }
/* 074 */ }
/* 075 */ if (!filter_isNull7) {
/* 076 */ filter_isNull6 = false; // resultCode could change nullability.
/* 077 */ filter_value6 = filter_value7 == 0;
/* 078 */
/* 079 */ }
/* 080 */ if (filter_isNull6 || !filter_value6) continue;
/* 081 */
/* 082 */ /* (input[1, int] <= 4) */
/* 083 */ boolean filter_value11 = false;
/* 084 */ filter_value11 = filter_value1 <= 4;
/* 085 */ if (!filter_value11) continue;
/* 086 */
/* 087 */ /* (input[1, int] > 1) */
/* 088 */ boolean filter_value14 = false;
/* 089 */ filter_value14 = filter_value1 > 1;
/* 090 */ if (!filter_value14) continue;
/* 091 */
/* 092 */ filter_metricValue.add(1);
/* 093 */
/* 094 */ /*** CONSUME: Project [_1#0 AS k#3,_2#1 AS v#4] */
/* 095 */
/* 096 */ /* input[0, string] */
/* 097 */ /* input[0, string] */
/* 098 */ boolean filter_isNull = inputadapter_row.isNullAt(0);
/* 099 */ UTF8String filter_value = filter_isNull ? null : (inputadapter_row.getUTF8String(0));
/* 100 */ project_holder.reset();
/* 101 */
/* 102 */ project_rowWriter.zeroOutNullBytes();
/* 103 */
/* 104 */ if (filter_isNull) {
/* 105 */ project_rowWriter.setNullAt(0);
/* 106 */ } else {
/* 107 */ project_rowWriter.write(0, filter_value);
/* 108 */ }
/* 109 */
/* 110 */ project_rowWriter.write(1, filter_value1);
/* 111 */ project_result.setTotalSize(project_holder.totalSize());
/* 112 */ append(project_result.copy());
/* 113 */ }
/* 114 */ }
/* 115 */ }
````
Generated code with this PR
````java
/* 032 */ protected void processNext() throws java.io.IOException {
/* 033 */ /*** PRODUCE: Project [_1#0 AS k#3,_2#1 AS v#4] */
/* 034 */
/* 035 */ /*** PRODUCE: Filter (((_2#1 % 2) = 0) && ((_2#1 <= 5) && (_2#1 > 1))) */
/* 036 */
/* 037 */ /*** PRODUCE: INPUT */
/* 038 */
/* 039 */ while (!shouldStop() && inputadapter_input.hasNext()) {
/* 040 */ InternalRow inputadapter_row = (InternalRow) inputadapter_input.next();
/* 041 */ /*** CONSUME: Filter (((_2#1 % 2) = 0) && ((_2#1 <= 5) && (_2#1 > 1))) */
/* 042 */ /* input[1, int] */
/* 043 */ int filter_value1 = inputadapter_row.getInt(1);
/* 044 */
/* 045 */ /* ((input[1, int] % 2) = 0) */
/* 046 */ /* (input[1, int] % 2) */
/* 047 */ int filter_value3 = (int)(filter_value1 % 2);
/* 048 */
/* 049 */ boolean filter_value2 = false;
/* 050 */ filter_value2 = filter_value3 == 0;
/* 051 */ if (!filter_value2) continue;
/* 052 */
/* 053 */ /* (input[1, int] <= 5) */
/* 054 */ boolean filter_value7 = false;
/* 055 */ filter_value7 = filter_value1 <= 5;
/* 056 */ if (!filter_value7) continue;
/* 057 */
/* 058 */ /* (input[1, int] > 1) */
/* 059 */ boolean filter_value10 = false;
/* 060 */ filter_value10 = filter_value1 > 1;
/* 061 */ if (!filter_value10) continue;
/* 062 */
/* 063 */ filter_metricValue.add(1);
/* 064 */
/* 065 */ /*** CONSUME: Project [_1#0 AS k#3,_2#1 AS v#4] */
/* 066 */
/* 067 */ /* input[0, string] */
/* 068 */ /* input[0, string] */
/* 069 */ boolean filter_isNull = inputadapter_row.isNullAt(0);
/* 070 */ UTF8String filter_value = filter_isNull ? null : (inputadapter_row.getUTF8String(0));
/* 071 */ project_holder.reset();
/* 072 */
/* 073 */ project_rowWriter.zeroOutNullBytes();
/* 074 */
/* 075 */ if (filter_isNull) {
/* 076 */ project_rowWriter.setNullAt(0);
/* 077 */ } else {
/* 078 */ project_rowWriter.write(0, filter_value);
/* 079 */ }
/* 080 */
/* 081 */ project_rowWriter.write(1, filter_value1);
/* 082 */ project_result.setTotalSize(project_holder.totalSize());
/* 083 */ append(project_result.copy());
/* 084 */ }
/* 085 */ }
/* 086 */ }
````
Author: Kazuaki Ishizaki <ishizaki@jp.ibm.com>
Closes #11684 from kiszk/SPARK-13844.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
proper
## What changes were proposed in this pull request?
The indentation of debug log output by `CodeGenerator` is weird.
The first line of the generated code should be put on the next line of the first line of the log message.
```
16/03/28 11:10:24 DEBUG CodeGenerator: /* 001 */
/* 002 */ public java.lang.Object generate(Object[] references) {
/* 003 */ return new SpecificSafeProjection(references);
...
```
After this patch is applied, we get debug log like as follows.
```
16/03/28 10:45:50 DEBUG CodeGenerator:
/* 001 */
/* 002 */ public java.lang.Object generate(Object[] references) {
/* 003 */ return new SpecificSafeProjection(references);
...
```
## How was this patch tested?
Ran some jobs and checked debug logs.
Author: Kousuke Saruta <sarutak@oss.nttdata.co.jp>
Closes #11990 from sarutak/fix-debuglog-indentation.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR fixes the following two testcases in order to test the correct usages.
```
checkSqlGeneration("SELECT substr('This is a test', 'is')")
checkSqlGeneration("SELECT substring('This is a test', 'is')")
```
Actually, the testcases works but tests on exceptional cases.
```
scala> sql("SELECT substr('This is a test', 'is')")
res0: org.apache.spark.sql.DataFrame = [substring(This is a test, CAST(is AS INT), 2147483647): string]
scala> sql("SELECT substr('This is a test', 'is')").collect()
res1: Array[org.apache.spark.sql.Row] = Array([null])
```
## How was this patch tested?
Pass the modified unit tests.
Author: Dongjoon Hyun <dongjoon@apache.org>
Closes #11963 from dongjoon-hyun/fix_substr_testcase.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
"Alter Database"
#### What changes were proposed in this pull request?
This PR is to provide native parsing support for two DDL commands: ```Describe Database``` and ```Alter Database Set Properties```
Based on the Hive DDL document:
https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL
##### 1. ALTER DATABASE
**Syntax:**
```SQL
ALTER (DATABASE|SCHEMA) database_name SET DBPROPERTIES (property_name=property_value, ...)
```
- `ALTER DATABASE` is to add new (key, value) pairs into `DBPROPERTIES`
##### 2. DESCRIBE DATABASE
**Syntax:**
```SQL
DESCRIBE DATABASE [EXTENDED] db_name
```
- `DESCRIBE DATABASE` shows the name of the database, its comment (if one has been set), and its root location on the filesystem. When `extended` is true, it also shows the database's properties
#### How was this patch tested?
Added the related test cases to `DDLCommandSuite`
Author: gatorsmile <gatorsmile@gmail.com>
Author: xiaoli <lixiao1983@gmail.com>
Author: Xiao Li <xiaoli@Xiaos-MacBook-Pro.local>
This patch had conflicts when merged, resolved by
Committer: Yin Huai <yhuai@databricks.com>
Closes #11977 from gatorsmile/parseAlterDatabase.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
JIRA: https://issues.apache.org/jira/browse/SPARK-14157
We only parse create function command. In order to support native drop function command, we need to parse it too.
From Hive [manual](https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL#LanguageManualDDL-Create/Drop/ReloadFunction), the drop function command has syntax as:
DROP [TEMPORARY] FUNCTION [IF EXISTS] function_name;
## How was this patch tested?
Added test into `DDLCommandSuite`.
Author: Liang-Chi Hsieh <simonh@tw.ibm.com>
Closes #11959 from viirya/parse-drop-func.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
## What changes were proposed in this pull request?
This PR implements `FileFormat.buildReader()` for our ORC data source. It also fixed several minor styling issues related to `HadoopFsRelation` planning code path.
Note that `OrcNewInputFormat` doesn't rely on `OrcNewSplit` for creating `OrcRecordReader`s, plain `FileSplit` is just fine. That's why we can simply create the record reader with the help of `OrcNewInputFormat` and `FileSplit`.
## How was this patch tested?
Existing test cases should do the work
Author: Cheng Lian <lian@databricks.com>
Closes #11936 from liancheng/spark-14116-build-reader-for-orc.
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
### What changes were proposed in this pull request?
Based on the Hive DDL document https://cwiki.apache.org/confluence/display/Hive/LanguageManual+DDL
The syntax of DDL command for Drop Database is
```SQL
DROP (DATABASE|SCHEMA) [IF EXISTS] database_name [RESTRICT|CASCADE];
```
- If `IF EXISTS` is not specified, the default behavior is to issue a warning message if `database_name` does't exist
- `RESTRICT` is the default behavior.
This PR is to provide a native parsing support for `DROP DATABASE`.
#### How was this patch tested?
Added a test case `DDLCommandSuite`
Author: gatorsmile <gatorsmile@gmail.com>
Closes #11962 from gatorsmile/parseDropDatabase.
|
