Yes, if you use totally the Calcite runtime, modify the queries is a way to
avoid huge memory consumption.
Best,
Danny Chan
在 2020年2月21日 +0800 PM3:52，Yang Liu <whilg...@gmail.com>，写道：
> Thanks Danny,
>
> multiplying row count and factor seems not help. In following query
>
> PLAN: EnumerableLimit(fetch=[10])
> EnumerableHashJoin(condition=[=($2, $8)], joinType=[inner])
> JdbcToEnumerableConverter
> JdbcTableScan(table=[[insight_user, user_tab]])
> ElasticsearchToEnumerableConverter
> ElasticsearchProject(status=[ITEM($0, 'status')], user=[ITEM($0,
> 'user')])
> ElasticsearchTableScan(table=[[es, es_table_1]])
> {10.0 rows, 828.9306334778566 cpu, 0.0 io}
>
> The main factor affecting performance is the ElasticsearchTableScan,
> although final row count is 10, intermediate result could be quite huge.
>
> If I am not wrong, when doing aggregation, Calcite provides two ways: in
> memory and in Spark. We just use the `in memory` option, so we care about
> the memory usage.
> If there is no internal solution I think we may need to parse the SQL
> first, if there is no limit or limit is too huge in the SQL, we reject the
> execution or rewrite the LIMIT part...
>
> Thanks
>
>
>
> Danny Chan <yuzhao....@gmail.com> 于2020年2月20日周四 下午10:22写道：
>
> > Sorry to tell that Calcite runtime is not designed to query bigdata. It
> > actually executes all the operator in single task.
> >
> > In distributed systems like Spark or Flink, we usually fallback to merge
> > join instead of hash when the data is huge.
> >
> > Danny Chan <yuzhao....@gmail.com>于2020年2月20日 周四下午10:16写道：
> >
> > > You are right, we have only row count into the source. There is no meta
> > > can describe the query performance. Can you multiply the row count with a
> > > explicit factor?
> > >
> > > Yang Liu <whilg...@gmail.com>于2020年2月20日 周四下午5:05写道：
> > >
> > > > Currently I can use following code to getCost,
> > > >
> > > > def getCost(rootSchema: SchemaPlus, sql: String): RelOptCost = {
> > > > val configBuilder = Frameworks.newConfigBuilder()
> > > > configBuilder.defaultSchema(rootSchema)
> > > > val frameworkConfig = configBuilder.build()
> > > > val planner = Frameworks.getPlanner(frameworkConfig)
> > > >
> > > > val sqlNode = planner.parse(sql)
> > > > val validate = planner.validate(sqlNode)
> > > > val rel = planner.rel(validate).project()
> > > > val mq = rel.getCluster().getMetadataQuery
> > > > mq.getNonCumulativeCost(rel)
> > > > }
> > > >
> > > > but seems not very correct, for example,
> > > >
> > > > between two mysql
> > > > PLAN: EnumerableCalc(expr#0..3=[{inputs}], proj#0..2=[{exprs}])
> > > > EnumerableHashJoin(condition=[=($0, $3)], joinType=[inner])
> > > > JdbcToEnumerableConverter
> > > > JdbcTableScan(table=[[perm, user_table]])
> > > > JdbcToEnumerableConverter
> > > > JdbcProject(id=[$0])
> > > > JdbcTableScan(table=[[insight_user, user_tab]])
> > > > {1500.0 rows, 4500.0 cpu, 0.0 io}
> > > >
> > > >
> > > > between es and mysql
> > > > PLAN: EnumerableLimit(fetch=[10])
> > > > EnumerableHashJoin(condition=[=($2, $8)], joinType=[inner])
> > > > JdbcToEnumerableConverter
> > > > JdbcTableScan(table=[[insight_user, user_tab]])
> > > > ElasticsearchToEnumerableConverter
> > > > ElasticsearchProject(status=[ITEM($0, 'status')], user=[ITEM($0,
> > > > 'user')])
> > > > ElasticsearchTableScan(table=[[es, es_table_1]])
> > > > {10.0 rows, 828.9306334778566 cpu, 0.0 io}
> > > >
> > > > in practice the later one is much slower than the first one since the es
> > > > table is very large.
> > > >
> > > > Seems the planner cost is a logical estimate, may I know the correct
> > > > usage of it?
> > > >
> > > > Thanks
> > > >
> > > > On 2020/02/20 03:00:47, Yang Liu <whilg...@gmail.com> wrote:
> > > > > Hi all,
> > > > >
> > > > > Dive into docs and I find the getJoinRowCount
> > > > > <
> > > >
> > https://calcite.apache.org/apidocs/org/apache/calcite/rel/metadata/RelMdUtil.html#getJoinRowCount(org.apache.calcite.rel.metadata.RelMetadataQuery,org.apache.calcite.rel.core.Join,org.apache.calcite.rex.RexNode)
> > > > >
> > > > > seems the very thing I am looking for, I can reject the join when the
> > > > rows
> > > > > are too many. Any ideas?
> > > > >
> > > > > Yang Liu <whilg...@gmail.com> 于2020年2月17日周一 下午5:13写道：
> > > > >
> > > > > > Thanks Muhammad, can help give more detailed description?
> > > > > > Currently I have searched a RefOptCost, is this the one you are
> > > > referring
> > > > > > to?
> > > > > > And I need to clarify, I do not mind the cost happened in
> > datasources
> > > > > > since that will not affect my application, may just take a longer
> > > > time to
> > > > > > get the result if the cost is high.
> > > > > > I care about the cost in our application and afraid of OOM.
> > > > > >
> > > > > > Thanks
> > > > > >
> > > > > > Muhammad Gelbana <mgelb...@apache.org> 于2020年2月17日周一 上午4:26写道：
> > > > > >
> > > > > > > If your only concern is about memory utilization, I would try
> > > > estimating
> > > > > > > this using the plan's cost. But I guess you'll have run some tests
> > to
> > > > > > > estimate the ranges you can accept.
> > > > > > >
> > > > > > >
> > > > > > > On Sun, Feb 16, 2020 at 5:50 PM Yang Liu <whilg...@gmail.com>
> > wrote:
> > > > > > >
> > > > > > > > Is it possible to have some limitations on the SQLs to make sure
> > > > our
> > > > > > > > application which depends on Calcite is "safe"? For example,
> > > > > > > > when
> > > > do
> > > > > > > merge
> > > > > > > > joining between 2 large datasources, our application maybe OOM
> > > > since the
> > > > > > > > joining process is in memory. If we have the "limitation
> > > > mechanism", we
> > > > > > > can
> > > > > > > > refuse to execute the joining to avoid OOM.
> > > > > > > >
> > > > > > > > Or we can only do the check outside Calcite?
> > > > > > > >
> > > > > > > > Thanks
> > > > > > > >
> > > > > > >
> > > > > >
> > > > >
> > > >
> > >
> >
