You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@calcite.apache.org by Shawn Weeks <sw...@weeksconsulting.us> on 2019/12/31 17:23:11 UTC
Understanding the Window Enumerable Process
Hi, I’m trying to troubleshoot an issue with the Apache NiFi projects use of Calcite to allow queries against flow files. NiFi presents an enumerable interface to Calcite org.apache.calcite.linq4j.EnumerableDefaults. For queries with simple where clause and aggregations everything works great but when I use window analytic functions like row_number against larger sets of records(2gb) the processes uses all memory allocated and every cpu core. I’ve attached a profiler and I’m only ever seeing 2-3 MB in live bytes and most of the CPU usage is NiFi parsing the records.
I was hoping someone could give me a high level overview of how Calcite is executing the window function. I would expect it to have to iterate through the data at least once for the window and once for the data itself.
Here is the query being executed and a trace of the Calcite execution plan.
select "antiNucleus",
"eventFile",
"eventNumber",
"eventTime",
"histFile",
"multiplicity",
"NaboveLb",
"NbelowLb",
"NLb",
"primaryTracks",
"prodTime",
"Pt",
"runNumber",
"vertexX",
"vertexY",
"vertexZ",
row_number() over() id
from flowfile;
7199 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - Root: rel#10:Subset#1.ENUMERABLE.[]
Original rel:
LogicalProject(subset=[rel#10:Subset#1.ENUMERABLE.[]], antiNucleus=[$0], eventFile=[$1], eventNumber=[$2], eventTime=[$3], histFile=[$4], multiplicity=[$5], NaboveLb=[$6], NbelowLb=[$7], NLb=[$8], primaryTracks=[$9], prodTime=[$10], Pt=[$11], runNumber=[$12], vertexX=[$13], vertexY=[$14], vertexZ=[$15], id=[ROW_NUMBER() OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)]): rowcount = 100.0, cumulative cost = {100.0 rows, 1700.0 cpu, 0.0 io}, id = 7
FlowFileTableScan(subset=[rel#6:Subset#0.ENUMERABLE.[]], table=[[FLOWFILE]], fields=[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]]): rowcount = 100.0, cumulative cost = {100.0 rows, 101.0 cpu, 0.0 io}, id = 0
Sets:
Set#0, type: RecordType(JavaType(class java.lang.String) antiNucleus, JavaType(class java.lang.String) eventFile, JavaType(class java.lang.String) eventNumber, JavaType(class java.lang.String) eventTime, JavaType(class java.lang.String) histFile, JavaType(class java.lang.String) multiplicity, JavaType(class java.lang.String) NaboveLb, JavaType(class java.lang.String) NbelowLb, JavaType(class java.lang.String) NLb, JavaType(class java.lang.String) primaryTracks, JavaType(class java.lang.String) prodTime, JavaType(class java.lang.String) Pt, JavaType(class java.lang.String) runNumber, JavaType(class java.lang.String) vertexX, JavaType(class java.lang.String) vertexY, JavaType(class java.lang.String) vertexZ)
rel#6:Subset#0.ENUMERABLE.[], best=rel#0, importance=0.81
rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]), rowcount=100.0, cumulative cost={100.0 rows, 101.0 cpu, 0.0 io}
Set#1, type: RecordType(JavaType(class java.lang.String) antiNucleus, JavaType(class java.lang.String) eventFile, JavaType(class java.lang.String) eventNumber, JavaType(class java.lang.String) eventTime, JavaType(class java.lang.String) histFile, JavaType(class java.lang.String) multiplicity, JavaType(class java.lang.String) NaboveLb, JavaType(class java.lang.String) NbelowLb, JavaType(class java.lang.String) NLb, JavaType(class java.lang.String) primaryTracks, JavaType(class java.lang.String) prodTime, JavaType(class java.lang.String) Pt, JavaType(class java.lang.String) runNumber, JavaType(class java.lang.String) vertexX, JavaType(class java.lang.String) vertexY, JavaType(class java.lang.String) vertexZ, BIGINT id)
rel#8:Subset#1.NONE.[], best=null, importance=0.9
rel#7:LogicalProject.NONE.[](input=rel#6:Subset#0.ENUMERABLE.[],antiNucleus=$0,eventFile=$1,eventNumber=$2,eventTime=$3,histFile=$4,multiplicity=$5,NaboveLb=$6,NbelowLb=$7,NLb=$8,primaryTracks=$9,prodTime=$10,Pt=$11,runNumber=$12,vertexX=$13,vertexY=$14,vertexZ=$15,id=ROW_NUMBER() OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)), rowcount=100.0, cumulative cost={inf}
rel#13:LogicalWindow.NONE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])), rowcount=100.0, cumulative cost={inf}
rel#10:Subset#1.ENUMERABLE.[], best=rel#15, importance=1.0
rel#11:AbstractConverter.ENUMERABLE.[](input=rel#8:Subset#1.NONE.[],convention=ENUMERABLE,sort=[]), rowcount=100.0, cumulative cost={inf}
rel#15:EnumerableWindow.ENUMERABLE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])), rowcount=100.0, cumulative cost={200.0 rows, 301.0 cpu, 0.0 io}
7200 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new EnumerableWindow#16
7215 [pool-1-thread-1] DEBUG org.apache.calcite.plan.RelOptPlanner - Cheapest plan:
EnumerableWindow(window#0=[window(partition {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])]): rowcount = 100.0, cumulative cost = {200.0 rows, 301.0 cpu, 0.0 io}, id = 16
FlowFileTableScan(table=[[FLOWFILE]], fields=[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]]): rowcount = 100.0, cumulative cost = {100.0 rows, 101.0 cpu, 0.0 io}, id = 0
7215 [pool-1-thread-1] DEBUG org.apache.calcite.plan.RelOptPlanner - Provenance:
EnumerableWindow#16
direct
rel#15:EnumerableWindow.ENUMERABLE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()]))
call#112 rule [EnumerableWindowRule]
rel#13:LogicalWindow.NONE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()]))
call#64 rule [ProjectToWindowRule:project]
rel#7:LogicalProject.NONE.[](input=rel#6:Subset#0.ENUMERABLE.[],antiNucleus=$0,eventFile=$1,eventNumber=$2,eventTime=$3,histFile=$4,multiplicity=$5,NaboveLb=$6,NbelowLb=$7,NLb=$8,primaryTracks=$9,prodTime=$10,Pt=$11,runNumber=$12,vertexX=$13,vertexY=$14,vertexZ=$15,id=ROW_NUMBER() OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW))
no parent
rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15])
no parent
7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new HepRelVertex#17
7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new EnumerableWindow#18
7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new HepRelVertex#19
7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner -
Breadth-first from root: {
HepRelVertex#19 = rel#18:EnumerableWindow.ENUMERABLE.[[]](input=HepRelVertex#17,window#0=window(partition {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])), rowcount=100.0, cumulative cost={200.0 rows, 301.0 cpu, 0.0 io}
HepRelVertex#17 = rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]), rowcount=100.0, cumulative cost={100.0 rows, 101.0 cpu, 0.0 io}
Re: Understanding the Window Enumerable Process
Posted by Feng Zhu <we...@gmail.com>.
The memory issue may be related with the implementation.
For row_number() function, the runtime processing logic is demonstrated as
below.
It can be seen that too many data structures (i.e., list/array/map) will be
created.
Best,
Feng
=====================================================================================================
*// 1.Loading data set into a List*
final java.util.List tempList = (java.util.List)
org.apache.calcite.linq4j.Linq4j.asEnumerable(***your data set***).into(new
java.util.ArrayList());
*// 2.Sorting the data set (Note: a new map&Array&Iterator will be
constructed, see SortedMultiMap)*
final java.util.Comparator comparator = new java.util.Comparator(){...};
final java.util.Iterator iterator =
org.apache.calcite.runtime.SortedMultiMap.singletonArrayIterator(comparator,
tempList);
*// 3.Constructing a new ArrayList for result*
final java.util.ArrayList _list = new
java.util.ArrayList(tempList.size());
*// 4. Computing row_number()*
Long a0w0 = (Long) null; *// for row_number*
while (iterator.hasNext()) {
final Object[] _rows = (Object[]) iterator.next();
for (int i = 0; i < _rows.length; ++i) {
final org.apache.calcite.test.JdbcTest.Employee row =
(org.apache.calcite.test.JdbcTest.Employee) _rows[i];
......
_list.add(new Object[] {
...... // row columns
a0w0});
}
}
tempList.clear();
final org.apache.calcite.linq4j.Enumerable _inputEnumerable =
org.apache.calcite.linq4j.Linq4j.asEnumerable(_list);
return new org.apache.calcite.linq4j.AbstractEnumerable(){
public org.apache.calcite.linq4j.Enumerator enumerator() {
return new org.apache.calcite.linq4j.Enumerator(){
public final org.apache.calcite.linq4j.Enumerator
inputEnumerator = _inputEnumerable.enumerator();
......
public Object current() {
final Object[] current = (Object[]) inputEnumerator.current();
return new Object[] { ... // row columns
};
}
};
}
};
Shawn Weeks <sw...@weeksconsulting.us> 于2020年1月3日周五 上午12:16写道：
> In my example I'm not setting a partition key as I'm trying to generate
> just a sequential list of integers. It sounds like everything is held in
> memory in a SortedMultiMap which explains the high memory usage though I'm
> not sure why it's using several times the memory as the input record set.
> My test file is a 2gb CSV and I'll hit out of memory with an '-Xmx16g' on
> the process.
>
> My last question was to try and figure out how many times Calcite would
> have to iterate over the data to produce a result as when I give it large
> amounts of memory it does eventually finish it just takes substantially
> longer than the time to iterate through the data once. For example a simple
> select * might take 2-3 minutes for my test file but the row_number version
> takes 30+ minutes if it ever finishes.
>
> Just in case I missed it, Calcite doesn't have a simple counter function
> like Oracles rownum function does it?
>
> Thanks
> Shawn
>
> On 1/1/20, 9:55 PM, "Feng Zhu" <we...@gmail.com> wrote:
>
> Hi, Shawn
>
> I was hoping someone could give me a high level overview of how
> Calcite is
> > executing the window function.
> >
> Calcite runtime generates Java code and complies the code with Janino.
> For
> Window Enumerable Process,
> there is a sample piece of code in
>
> *org.apache.calcite.adapter.enumerable.EnumerableWindow#sampleOfTheGeneratedWindowedAggregate()*,
> which may be what you need.
>
> In general, input rows will be firstly organized into a*
> org.apache.calcite.runtime.SortedMultiMap* according to partitioning
> key,
> and then all of the windowed aggregate functions will be evaluated for
> each
> list of rows that have the same partitioning key.
>
> I would expect it to have to iterate through the data at least once
> for the
> > window and once for the data itself.
> >
> Could you elaborate?
>
> Best,
> Feng
>
> Shawn Weeks <sw...@weeksconsulting.us> 于2020年1月1日周三 上午1:23写道：
>
> > Hi, I’m trying to troubleshoot an issue with the Apache NiFi
> projects use
> > of Calcite to allow queries against flow files. NiFi presents an
> enumerable
> > interface to Calcite org.apache.calcite.linq4j.EnumerableDefaults.
> For
> > queries with simple where clause and aggregations everything works
> great
> > but when I use window analytic functions like row_number against
> larger
> > sets of records(2gb) the processes uses all memory allocated and
> every cpu
> > core. I’ve attached a profiler and I’m only ever seeing 2-3 MB in
> live
> > bytes and most of the CPU usage is NiFi parsing the records.
> >
> > I was hoping someone could give me a high level overview of how
> Calcite is
> > executing the window function. I would expect it to have to iterate
> through
> > the data at least once for the window and once for the data itself.
> >
> > Here is the query being executed and a trace of the Calcite
> execution plan.
> >
> > select "antiNucleus",
> > "eventFile",
> > "eventNumber",
> > "eventTime",
> > "histFile",
> > "multiplicity",
> > "NaboveLb",
> > "NbelowLb",
> > "NLb",
> > "primaryTracks",
> > "prodTime",
> > "Pt",
> > "runNumber",
> > "vertexX",
> > "vertexY",
> > "vertexZ",
> > row_number() over() id
> > from flowfile;
> >
> > 7199 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner -
> Root:
> > rel#10:Subset#1.ENUMERABLE.[]
> > Original rel:
> > LogicalProject(subset=[rel#10:Subset#1.ENUMERABLE.[]],
> antiNucleus=[$0],
> > eventFile=[$1], eventNumber=[$2], eventTime=[$3], histFile=[$4],
> > multiplicity=[$5], NaboveLb=[$6], NbelowLb=[$7], NLb=[$8],
> > primaryTracks=[$9], prodTime=[$10], Pt=[$11], runNumber=[$12],
> > vertexX=[$13], vertexY=[$14], vertexZ=[$15], id=[ROW_NUMBER() OVER
> (ROWS
> > BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)]): rowcount = 100.0,
> > cumulative cost = {100.0 rows, 1700.0 cpu, 0.0 io}, id = 7
> > FlowFileTableScan(subset=[rel#6:Subset#0.ENUMERABLE.[]],
> > table=[[FLOWFILE]], fields=[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11,
> 12, 13,
> > 14, 15]]): rowcount = 100.0, cumulative cost = {100.0 rows, 101.0
> cpu, 0.0
> > io}, id = 0
> >
> > Sets:
> > Set#0, type: RecordType(JavaType(class java.lang.String) antiNucleus,
> > JavaType(class java.lang.String) eventFile, JavaType(class
> > java.lang.String) eventNumber, JavaType(class java.lang.String)
> eventTime,
> > JavaType(class java.lang.String) histFile, JavaType(class
> java.lang.String)
> > multiplicity, JavaType(class java.lang.String) NaboveLb,
> JavaType(class
> > java.lang.String) NbelowLb, JavaType(class java.lang.String) NLb,
> > JavaType(class java.lang.String) primaryTracks, JavaType(class
> > java.lang.String) prodTime, JavaType(class java.lang.String) Pt,
> > JavaType(class java.lang.String) runNumber, JavaType(class
> > java.lang.String) vertexX, JavaType(class java.lang.String) vertexY,
> > JavaType(class java.lang.String) vertexZ)
> > rel#6:Subset#0.ENUMERABLE.[], best=rel#0, importance=0.81
> >
> > rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1,
> 2, 3,
> > 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]), rowcount=100.0,
> cumulative
> > cost={100.0 rows, 101.0 cpu, 0.0 io}
> > Set#1, type: RecordType(JavaType(class java.lang.String) antiNucleus,
> > JavaType(class java.lang.String) eventFile, JavaType(class
> > java.lang.String) eventNumber, JavaType(class java.lang.String)
> eventTime,
> > JavaType(class java.lang.String) histFile, JavaType(class
> java.lang.String)
> > multiplicity, JavaType(class java.lang.String) NaboveLb,
> JavaType(class
> > java.lang.String) NbelowLb, JavaType(class java.lang.String) NLb,
> > JavaType(class java.lang.String) primaryTracks, JavaType(class
> > java.lang.String) prodTime, JavaType(class java.lang.String) Pt,
> > JavaType(class java.lang.String) runNumber, JavaType(class
> > java.lang.String) vertexX, JavaType(class java.lang.String) vertexY,
> > JavaType(class java.lang.String) vertexZ, BIGINT id)
> > rel#8:Subset#1.NONE.[], best=null, importance=0.9
> >
> >
> rel#7:LogicalProject.NONE.[](input=rel#6:Subset#0.ENUMERABLE.[],antiNucleus=$0,eventFile=$1,eventNumber=$2,eventTime=$3,histFile=$4,multiplicity=$5,NaboveLb=$6,NbelowLb=$7,NLb=$8,primaryTracks=$9,prodTime=$10,Pt=$11,runNumber=$12,vertexX=$13,vertexY=$14,vertexZ=$15,id=ROW_NUMBER()
> > OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)),
> rowcount=100.0,
> > cumulative cost={inf}
> >
> >
> rel#13:LogicalWindow.NONE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> > {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> > [ROW_NUMBER()])), rowcount=100.0, cumulative cost={inf}
> > rel#10:Subset#1.ENUMERABLE.[], best=rel#15, importance=1.0
> >
> >
> rel#11:AbstractConverter.ENUMERABLE.[](input=rel#8:Subset#1.NONE.[],convention=ENUMERABLE,sort=[]),
> > rowcount=100.0, cumulative cost={inf}
> >
> >
> rel#15:EnumerableWindow.ENUMERABLE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> > {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> > [ROW_NUMBER()])), rowcount=100.0, cumulative cost={200.0 rows, 301.0
> cpu,
> > 0.0 io}
> >
> >
> > 7200 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner -
> new
> > EnumerableWindow#16
> > 7215 [pool-1-thread-1] DEBUG org.apache.calcite.plan.RelOptPlanner -
> > Cheapest plan:
> > EnumerableWindow(window#0=[window(partition {} order by [] rows
> between
> > UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])]): rowcount
> =
> > 100.0, cumulative cost = {200.0 rows, 301.0 cpu, 0.0 io}, id = 16
> > FlowFileTableScan(table=[[FLOWFILE]], fields=[[0, 1, 2, 3, 4, 5,
> 6, 7,
> > 8, 9, 10, 11, 12, 13, 14, 15]]): rowcount = 100.0, cumulative cost =
> {100.0
> > rows, 101.0 cpu, 0.0 io}, id = 0
> >
> > 7215 [pool-1-thread-1] DEBUG org.apache.calcite.plan.RelOptPlanner -
> > Provenance:
> > EnumerableWindow#16
> > direct
> >
> >
> rel#15:EnumerableWindow.ENUMERABLE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> > {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> > [ROW_NUMBER()]))
> > call#112 rule [EnumerableWindowRule]
> >
> >
> rel#13:LogicalWindow.NONE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> > {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> > [ROW_NUMBER()]))
> > call#64 rule [ProjectToWindowRule:project]
> >
> >
> rel#7:LogicalProject.NONE.[](input=rel#6:Subset#0.ENUMERABLE.[],antiNucleus=$0,eventFile=$1,eventNumber=$2,eventTime=$3,histFile=$4,multiplicity=$5,NaboveLb=$6,NbelowLb=$7,NLb=$8,primaryTracks=$9,prodTime=$10,Pt=$11,runNumber=$12,vertexX=$13,vertexY=$14,vertexZ=$15,id=ROW_NUMBER()
> > OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW))
> > no parent
> > rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1,
> 2, 3,
> > 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15])
> > no parent
> >
> > 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner -
> new
> > HepRelVertex#17
> > 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner -
> new
> > EnumerableWindow#18
> > 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner -
> new
> > HepRelVertex#19
> > 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner -
> > Breadth-first from root: {
> > HepRelVertex#19 =
> >
> rel#18:EnumerableWindow.ENUMERABLE.[[]](input=HepRelVertex#17,window#0=window(partition
> > {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> > [ROW_NUMBER()])), rowcount=100.0, cumulative cost={200.0 rows, 301.0
> cpu,
> > 0.0 io}
> > HepRelVertex#17 =
> > rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1,
> 2, 3,
> > 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]), rowcount=100.0,
> cumulative
> > cost={100.0 rows, 101.0 cpu, 0.0 io}
> >
>
>
>
Re: Understanding the Window Enumerable Process
Posted by Shawn Weeks <sw...@weeksconsulting.us>.
In my example I'm not setting a partition key as I'm trying to generate just a sequential list of integers. It sounds like everything is held in memory in a SortedMultiMap which explains the high memory usage though I'm not sure why it's using several times the memory as the input record set. My test file is a 2gb CSV and I'll hit out of memory with an '-Xmx16g' on the process.
My last question was to try and figure out how many times Calcite would have to iterate over the data to produce a result as when I give it large amounts of memory it does eventually finish it just takes substantially longer than the time to iterate through the data once. For example a simple select * might take 2-3 minutes for my test file but the row_number version takes 30+ minutes if it ever finishes.
Just in case I missed it, Calcite doesn't have a simple counter function like Oracles rownum function does it?
Thanks
Shawn
On 1/1/20, 9:55 PM, "Feng Zhu" <we...@gmail.com> wrote:
Hi, Shawn
I was hoping someone could give me a high level overview of how Calcite is
> executing the window function.
>
Calcite runtime generates Java code and complies the code with Janino. For
Window Enumerable Process,
there is a sample piece of code in
*org.apache.calcite.adapter.enumerable.EnumerableWindow#sampleOfTheGeneratedWindowedAggregate()*,
which may be what you need.
In general, input rows will be firstly organized into a*
org.apache.calcite.runtime.SortedMultiMap* according to partitioning key,
and then all of the windowed aggregate functions will be evaluated for each
list of rows that have the same partitioning key.
I would expect it to have to iterate through the data at least once for the
> window and once for the data itself.
>
Could you elaborate?
Best,
Feng
Shawn Weeks <sw...@weeksconsulting.us> 于2020年1月1日周三 上午1:23写道：
> Hi, I’m trying to troubleshoot an issue with the Apache NiFi projects use
> of Calcite to allow queries against flow files. NiFi presents an enumerable
> interface to Calcite org.apache.calcite.linq4j.EnumerableDefaults. For
> queries with simple where clause and aggregations everything works great
> but when I use window analytic functions like row_number against larger
> sets of records(2gb) the processes uses all memory allocated and every cpu
> core. I’ve attached a profiler and I’m only ever seeing 2-3 MB in live
> bytes and most of the CPU usage is NiFi parsing the records.
>
> I was hoping someone could give me a high level overview of how Calcite is
> executing the window function. I would expect it to have to iterate through
> the data at least once for the window and once for the data itself.
>
> Here is the query being executed and a trace of the Calcite execution plan.
>
> select "antiNucleus",
> "eventFile",
> "eventNumber",
> "eventTime",
> "histFile",
> "multiplicity",
> "NaboveLb",
> "NbelowLb",
> "NLb",
> "primaryTracks",
> "prodTime",
> "Pt",
> "runNumber",
> "vertexX",
> "vertexY",
> "vertexZ",
> row_number() over() id
> from flowfile;
>
> 7199 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - Root:
> rel#10:Subset#1.ENUMERABLE.[]
> Original rel:
> LogicalProject(subset=[rel#10:Subset#1.ENUMERABLE.[]], antiNucleus=[$0],
> eventFile=[$1], eventNumber=[$2], eventTime=[$3], histFile=[$4],
> multiplicity=[$5], NaboveLb=[$6], NbelowLb=[$7], NLb=[$8],
> primaryTracks=[$9], prodTime=[$10], Pt=[$11], runNumber=[$12],
> vertexX=[$13], vertexY=[$14], vertexZ=[$15], id=[ROW_NUMBER() OVER (ROWS
> BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)]): rowcount = 100.0,
> cumulative cost = {100.0 rows, 1700.0 cpu, 0.0 io}, id = 7
> FlowFileTableScan(subset=[rel#6:Subset#0.ENUMERABLE.[]],
> table=[[FLOWFILE]], fields=[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13,
> 14, 15]]): rowcount = 100.0, cumulative cost = {100.0 rows, 101.0 cpu, 0.0
> io}, id = 0
>
> Sets:
> Set#0, type: RecordType(JavaType(class java.lang.String) antiNucleus,
> JavaType(class java.lang.String) eventFile, JavaType(class
> java.lang.String) eventNumber, JavaType(class java.lang.String) eventTime,
> JavaType(class java.lang.String) histFile, JavaType(class java.lang.String)
> multiplicity, JavaType(class java.lang.String) NaboveLb, JavaType(class
> java.lang.String) NbelowLb, JavaType(class java.lang.String) NLb,
> JavaType(class java.lang.String) primaryTracks, JavaType(class
> java.lang.String) prodTime, JavaType(class java.lang.String) Pt,
> JavaType(class java.lang.String) runNumber, JavaType(class
> java.lang.String) vertexX, JavaType(class java.lang.String) vertexY,
> JavaType(class java.lang.String) vertexZ)
> rel#6:Subset#0.ENUMERABLE.[], best=rel#0, importance=0.81
>
> rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1, 2, 3,
> 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]), rowcount=100.0, cumulative
> cost={100.0 rows, 101.0 cpu, 0.0 io}
> Set#1, type: RecordType(JavaType(class java.lang.String) antiNucleus,
> JavaType(class java.lang.String) eventFile, JavaType(class
> java.lang.String) eventNumber, JavaType(class java.lang.String) eventTime,
> JavaType(class java.lang.String) histFile, JavaType(class java.lang.String)
> multiplicity, JavaType(class java.lang.String) NaboveLb, JavaType(class
> java.lang.String) NbelowLb, JavaType(class java.lang.String) NLb,
> JavaType(class java.lang.String) primaryTracks, JavaType(class
> java.lang.String) prodTime, JavaType(class java.lang.String) Pt,
> JavaType(class java.lang.String) runNumber, JavaType(class
> java.lang.String) vertexX, JavaType(class java.lang.String) vertexY,
> JavaType(class java.lang.String) vertexZ, BIGINT id)
> rel#8:Subset#1.NONE.[], best=null, importance=0.9
>
> rel#7:LogicalProject.NONE.[](input=rel#6:Subset#0.ENUMERABLE.[],antiNucleus=$0,eventFile=$1,eventNumber=$2,eventTime=$3,histFile=$4,multiplicity=$5,NaboveLb=$6,NbelowLb=$7,NLb=$8,primaryTracks=$9,prodTime=$10,Pt=$11,runNumber=$12,vertexX=$13,vertexY=$14,vertexZ=$15,id=ROW_NUMBER()
> OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)), rowcount=100.0,
> cumulative cost={inf}
>
> rel#13:LogicalWindow.NONE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> [ROW_NUMBER()])), rowcount=100.0, cumulative cost={inf}
> rel#10:Subset#1.ENUMERABLE.[], best=rel#15, importance=1.0
>
> rel#11:AbstractConverter.ENUMERABLE.[](input=rel#8:Subset#1.NONE.[],convention=ENUMERABLE,sort=[]),
> rowcount=100.0, cumulative cost={inf}
>
> rel#15:EnumerableWindow.ENUMERABLE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> [ROW_NUMBER()])), rowcount=100.0, cumulative cost={200.0 rows, 301.0 cpu,
> 0.0 io}
>
>
> 7200 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new
> EnumerableWindow#16
> 7215 [pool-1-thread-1] DEBUG org.apache.calcite.plan.RelOptPlanner -
> Cheapest plan:
> EnumerableWindow(window#0=[window(partition {} order by [] rows between
> UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])]): rowcount =
> 100.0, cumulative cost = {200.0 rows, 301.0 cpu, 0.0 io}, id = 16
> FlowFileTableScan(table=[[FLOWFILE]], fields=[[0, 1, 2, 3, 4, 5, 6, 7,
> 8, 9, 10, 11, 12, 13, 14, 15]]): rowcount = 100.0, cumulative cost = {100.0
> rows, 101.0 cpu, 0.0 io}, id = 0
>
> 7215 [pool-1-thread-1] DEBUG org.apache.calcite.plan.RelOptPlanner -
> Provenance:
> EnumerableWindow#16
> direct
>
> rel#15:EnumerableWindow.ENUMERABLE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> [ROW_NUMBER()]))
> call#112 rule [EnumerableWindowRule]
>
> rel#13:LogicalWindow.NONE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> [ROW_NUMBER()]))
> call#64 rule [ProjectToWindowRule:project]
>
> rel#7:LogicalProject.NONE.[](input=rel#6:Subset#0.ENUMERABLE.[],antiNucleus=$0,eventFile=$1,eventNumber=$2,eventTime=$3,histFile=$4,multiplicity=$5,NaboveLb=$6,NbelowLb=$7,NLb=$8,primaryTracks=$9,prodTime=$10,Pt=$11,runNumber=$12,vertexX=$13,vertexY=$14,vertexZ=$15,id=ROW_NUMBER()
> OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW))
> no parent
> rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1, 2, 3,
> 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15])
> no parent
>
> 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new
> HepRelVertex#17
> 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new
> EnumerableWindow#18
> 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new
> HepRelVertex#19
> 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner -
> Breadth-first from root: {
> HepRelVertex#19 =
> rel#18:EnumerableWindow.ENUMERABLE.[[]](input=HepRelVertex#17,window#0=window(partition
> {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> [ROW_NUMBER()])), rowcount=100.0, cumulative cost={200.0 rows, 301.0 cpu,
> 0.0 io}
> HepRelVertex#17 =
> rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1, 2, 3,
> 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]), rowcount=100.0, cumulative
> cost={100.0 rows, 101.0 cpu, 0.0 io}
>
Re: Understanding the Window Enumerable Process
Posted by Feng Zhu <we...@gmail.com>.
Hi, Shawn
I was hoping someone could give me a high level overview of how Calcite is
> executing the window function.
>
Calcite runtime generates Java code and complies the code with Janino. For
Window Enumerable Process,
there is a sample piece of code in
*org.apache.calcite.adapter.enumerable.EnumerableWindow#sampleOfTheGeneratedWindowedAggregate()*,
which may be what you need.
In general, input rows will be firstly organized into a*
org.apache.calcite.runtime.SortedMultiMap* according to partitioning key,
and then all of the windowed aggregate functions will be evaluated for each
list of rows that have the same partitioning key.
I would expect it to have to iterate through the data at least once for the
> window and once for the data itself.
>
Could you elaborate?
Best,
Feng
Shawn Weeks <sw...@weeksconsulting.us> 于2020年1月1日周三 上午1:23写道：
> Hi, I’m trying to troubleshoot an issue with the Apache NiFi projects use
> of Calcite to allow queries against flow files. NiFi presents an enumerable
> interface to Calcite org.apache.calcite.linq4j.EnumerableDefaults. For
> queries with simple where clause and aggregations everything works great
> but when I use window analytic functions like row_number against larger
> sets of records(2gb) the processes uses all memory allocated and every cpu
> core. I’ve attached a profiler and I’m only ever seeing 2-3 MB in live
> bytes and most of the CPU usage is NiFi parsing the records.
>
> I was hoping someone could give me a high level overview of how Calcite is
> executing the window function. I would expect it to have to iterate through
> the data at least once for the window and once for the data itself.
>
> Here is the query being executed and a trace of the Calcite execution plan.
>
> select "antiNucleus",
> "eventFile",
> "eventNumber",
> "eventTime",
> "histFile",
> "multiplicity",
> "NaboveLb",
> "NbelowLb",
> "NLb",
> "primaryTracks",
> "prodTime",
> "Pt",
> "runNumber",
> "vertexX",
> "vertexY",
> "vertexZ",
> row_number() over() id
> from flowfile;
>
> 7199 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - Root:
> rel#10:Subset#1.ENUMERABLE.[]
> Original rel:
> LogicalProject(subset=[rel#10:Subset#1.ENUMERABLE.[]], antiNucleus=[$0],
> eventFile=[$1], eventNumber=[$2], eventTime=[$3], histFile=[$4],
> multiplicity=[$5], NaboveLb=[$6], NbelowLb=[$7], NLb=[$8],
> primaryTracks=[$9], prodTime=[$10], Pt=[$11], runNumber=[$12],
> vertexX=[$13], vertexY=[$14], vertexZ=[$15], id=[ROW_NUMBER() OVER (ROWS
> BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)]): rowcount = 100.0,
> cumulative cost = {100.0 rows, 1700.0 cpu, 0.0 io}, id = 7
> FlowFileTableScan(subset=[rel#6:Subset#0.ENUMERABLE.[]],
> table=[[FLOWFILE]], fields=[[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13,
> 14, 15]]): rowcount = 100.0, cumulative cost = {100.0 rows, 101.0 cpu, 0.0
> io}, id = 0
>
> Sets:
> Set#0, type: RecordType(JavaType(class java.lang.String) antiNucleus,
> JavaType(class java.lang.String) eventFile, JavaType(class
> java.lang.String) eventNumber, JavaType(class java.lang.String) eventTime,
> JavaType(class java.lang.String) histFile, JavaType(class java.lang.String)
> multiplicity, JavaType(class java.lang.String) NaboveLb, JavaType(class
> java.lang.String) NbelowLb, JavaType(class java.lang.String) NLb,
> JavaType(class java.lang.String) primaryTracks, JavaType(class
> java.lang.String) prodTime, JavaType(class java.lang.String) Pt,
> JavaType(class java.lang.String) runNumber, JavaType(class
> java.lang.String) vertexX, JavaType(class java.lang.String) vertexY,
> JavaType(class java.lang.String) vertexZ)
> rel#6:Subset#0.ENUMERABLE.[], best=rel#0, importance=0.81
>
> rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1, 2, 3,
> 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]), rowcount=100.0, cumulative
> cost={100.0 rows, 101.0 cpu, 0.0 io}
> Set#1, type: RecordType(JavaType(class java.lang.String) antiNucleus,
> JavaType(class java.lang.String) eventFile, JavaType(class
> java.lang.String) eventNumber, JavaType(class java.lang.String) eventTime,
> JavaType(class java.lang.String) histFile, JavaType(class java.lang.String)
> multiplicity, JavaType(class java.lang.String) NaboveLb, JavaType(class
> java.lang.String) NbelowLb, JavaType(class java.lang.String) NLb,
> JavaType(class java.lang.String) primaryTracks, JavaType(class
> java.lang.String) prodTime, JavaType(class java.lang.String) Pt,
> JavaType(class java.lang.String) runNumber, JavaType(class
> java.lang.String) vertexX, JavaType(class java.lang.String) vertexY,
> JavaType(class java.lang.String) vertexZ, BIGINT id)
> rel#8:Subset#1.NONE.[], best=null, importance=0.9
>
> rel#7:LogicalProject.NONE.[](input=rel#6:Subset#0.ENUMERABLE.[],antiNucleus=$0,eventFile=$1,eventNumber=$2,eventTime=$3,histFile=$4,multiplicity=$5,NaboveLb=$6,NbelowLb=$7,NLb=$8,primaryTracks=$9,prodTime=$10,Pt=$11,runNumber=$12,vertexX=$13,vertexY=$14,vertexZ=$15,id=ROW_NUMBER()
> OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW)), rowcount=100.0,
> cumulative cost={inf}
>
> rel#13:LogicalWindow.NONE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> [ROW_NUMBER()])), rowcount=100.0, cumulative cost={inf}
> rel#10:Subset#1.ENUMERABLE.[], best=rel#15, importance=1.0
>
> rel#11:AbstractConverter.ENUMERABLE.[](input=rel#8:Subset#1.NONE.[],convention=ENUMERABLE,sort=[]),
> rowcount=100.0, cumulative cost={inf}
>
> rel#15:EnumerableWindow.ENUMERABLE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> [ROW_NUMBER()])), rowcount=100.0, cumulative cost={200.0 rows, 301.0 cpu,
> 0.0 io}
>
>
> 7200 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new
> EnumerableWindow#16
> 7215 [pool-1-thread-1] DEBUG org.apache.calcite.plan.RelOptPlanner -
> Cheapest plan:
> EnumerableWindow(window#0=[window(partition {} order by [] rows between
> UNBOUNDED PRECEDING and CURRENT ROW aggs [ROW_NUMBER()])]): rowcount =
> 100.0, cumulative cost = {200.0 rows, 301.0 cpu, 0.0 io}, id = 16
> FlowFileTableScan(table=[[FLOWFILE]], fields=[[0, 1, 2, 3, 4, 5, 6, 7,
> 8, 9, 10, 11, 12, 13, 14, 15]]): rowcount = 100.0, cumulative cost = {100.0
> rows, 101.0 cpu, 0.0 io}, id = 0
>
> 7215 [pool-1-thread-1] DEBUG org.apache.calcite.plan.RelOptPlanner -
> Provenance:
> EnumerableWindow#16
> direct
>
> rel#15:EnumerableWindow.ENUMERABLE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> [ROW_NUMBER()]))
> call#112 rule [EnumerableWindowRule]
>
> rel#13:LogicalWindow.NONE.[[]](input=rel#6:Subset#0.ENUMERABLE.[],window#0=window(partition
> {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> [ROW_NUMBER()]))
> call#64 rule [ProjectToWindowRule:project]
>
> rel#7:LogicalProject.NONE.[](input=rel#6:Subset#0.ENUMERABLE.[],antiNucleus=$0,eventFile=$1,eventNumber=$2,eventTime=$3,histFile=$4,multiplicity=$5,NaboveLb=$6,NbelowLb=$7,NLb=$8,primaryTracks=$9,prodTime=$10,Pt=$11,runNumber=$12,vertexX=$13,vertexY=$14,vertexZ=$15,id=ROW_NUMBER()
> OVER (ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW))
> no parent
> rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1, 2, 3,
> 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15])
> no parent
>
> 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new
> HepRelVertex#17
> 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new
> EnumerableWindow#18
> 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner - new
> HepRelVertex#19
> 7215 [pool-1-thread-1] TRACE org.apache.calcite.plan.RelOptPlanner -
> Breadth-first from root: {
> HepRelVertex#19 =
> rel#18:EnumerableWindow.ENUMERABLE.[[]](input=HepRelVertex#17,window#0=window(partition
> {} order by [] rows between UNBOUNDED PRECEDING and CURRENT ROW aggs
> [ROW_NUMBER()])), rowcount=100.0, cumulative cost={200.0 rows, 301.0 cpu,
> 0.0 io}
> HepRelVertex#17 =
> rel#0:FlowFileTableScan.ENUMERABLE.[](table=[FLOWFILE],fields=[0, 1, 2, 3,
> 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15]), rowcount=100.0, cumulative
> cost={100.0 rows, 101.0 cpu, 0.0 io}
>