raminqaf commented on code in PR #29837:
URL: https://github.com/apache/flink/pull/29837#discussion_r4525522678
##########
docs/content/docs/dev/table/tableApi.md:
##########
@@ -74,39 +74,6 @@ Table counts = orders
counts.execute().print();
```
-{{< /tab >}}
-{{< tab "Scala" >}}
-
-The Scala Table API is enabled by importing `org.apache.flink.table.api._`,
`org.apache.flink.api.scala._`, and `org.apache.flink.table.api.bridge.scala._`
(for bridging to/from DataStream).
-
-The following example shows how a Scala Table API program is constructed.
Table fields are referenced using Scala's String interpolation using a dollar
character (`$`).
-
-```scala
-import org.apache.flink.api.scala._
-import org.apache.flink.table.api._
-import org.apache.flink.table.api.bridge.scala._
-
-// environment configuration
-val settings = EnvironmentSettings
- .newInstance()
- .inStreamingMode()
- .build()
-
-val tEnv = TableEnvironment.create(settings)
-
-// register Orders table in table environment
-// ...
-
-// specify table program
-val orders = tEnv.from("Orders") // schema (a, b, c, rowtime)
-
-val result = orders
- .groupBy($"a")
- .select($"a", $"b".count as "cnt")
- .execute()
- .print()
-```
-
{{< /tab >}}
{{< tab "Python" >}}
Review Comment:
This group lost its Scala tab, but the tab only used Table API calls. Its
one stale line was `import org.apache.flink.api.scala._`, and that package is
gone. The commit message says Scala Table API examples without the DataStream
API stay, and common.md keeps its Scala tab and only drops that import. Could
we restore [the
tab](https://github.com/apache/flink/blob/a21fa009bc5064f81525f501dea7da4759adbfc7/docs/content/docs/dev/table/tableApi.md#L78-L110)
without that import, here and in its lead-in sentence? If it should go
instead, [lines 41 and
43](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/dev/table/tableApi.md#L41-L43)
still promise a Scala example. The zh page has the same at [lines 40 and
42](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/dev/table/tableApi.md#L40-L42).
##########
docs/content/docs/dev/datastream/overview.md:
##########
@@ -260,10 +260,9 @@ Data Sources
{{< tab "Java" >}}
Sources are where your program reads its input from. You can attach a source
to your program by
-using `StreamExecutionEnvironment.addSource(sourceFunction)`. Flink comes with
a number of pre-implemented
-source functions, but you can always write your own custom sources by
implementing the `SourceFunction`
-for non-parallel sources, or by implementing the `ParallelSourceFunction`
interface or extending the
-`RichParallelSourceFunction` for parallel sources.
+using `StreamExecutionEnvironment.fromSource(source, watermarkStrategy,
sourceName)`. Flink comes with a
+number of pre-implemented sources, but you can always write your own custom
source by implementing the
+[Source API]({{< ref "docs/internals/sources" >}}).
Review Comment:
The new intro says to attach sources with `fromSource(...)` and the Source
API. The Custom bullet at [lines
307-308](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/dev/datastream/overview.md#L307-L308)
still recommends `addSource(new FlinkKafkaConsumer<>(...))`, and [line
432](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/dev/datastream/overview.md#L432)
calls `env.addSource(...)`. In the collection snippet, [line
458](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/dev/datastream/overview.md#L458)
keeps `fromCollection(data)`, while the zh page already has
[`fromData(data)`](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/dev/datastream/overview.md#L390).
Could we finish these on this page?
Two more Java snippets in files this PR edits still fail javac:
-
[tableApi.md:2080](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/dev/table/tableApi.md#L2080)
has `.orderBy("proctime")`. `OverWindowPartitioned` only has
`orderBy(Expression)`, so this should be `orderBy($("proctime"))` like the line
above. Same in
[zh](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/dev/table/tableApi.md#L2081).
-
[catalogs.md:645](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/sql/catalogs.md#L645)
calls `registerCatalog(new CustomCatalog("myCatalog"))`. Only
`registerCatalog(String, Catalog)` exists, and it is deprecated.
`createCatalog` with a `CatalogDescriptor` would match the other snippets in
this file. Same in
[zh](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/sql/catalogs.md#L640).
##########
docs/content/docs/sql/timezone.md:
##########
@@ -163,9 +165,9 @@ Flink SQL> DESC MyView1;
| CURRENT_DATE | DATE | false | |
| |
| CURRENT_TIME | TIME(0) | false | |
| |
| CURRENT_TIMESTAMP | TIMESTAMP_LTZ(3) | false | |
| |
-|CURRENT_ROW_TIMESTAMP() | TIMESTAMP_LTZ(3) | false | |
| |
-| NOW() | TIMESTAMP_LTZ(3) | false | |
| |
-| PROCTIME() | TIMESTAMP_LTZ(3) *PROCTIME* | false | |
| |
+| EXPR$5 | TIMESTAMP_LTZ(3) | false | |
| |
+| EXPR$6 | TIMESTAMP_LTZ(3) | false | |
| |
+| EXPR$7 | TIMESTAMP_LTZ(3) *PROCTIME* | false | |
| |
Review Comment:
DESC now lists `EXPR$5`, `EXPR$6` and `EXPR$7`. The two `SELECT * FROM
MyView1` outputs below still use `CURRENT_ROW_TIMESTAMP()`, `NOW()` and
`PROCTIME()` as headers, at [line
181](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/sql/timezone.md#L181)
and [line
194](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/sql/timezone.md#L194).
On master that SELECT fails with FLINK-41409, so those tables are not real
output. Could we give every function in the `CREATE VIEW` an alias? Then the
example runs, and DESC and both SELECT outputs show the same names. The zh page
has the same headers at [lines 181 and
194](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/sql/timezone.md#L181-L194).
##########
docs/content/docs/sql/reference/queries/overview.md:
##########
@@ -142,16 +102,16 @@ result = table_env \
.sql_query("SELECT SUM(amount) FROM %s WHERE product LIKE '%%Rubber%%'" %
table)
# create and register a TableSink
-schema = Schema.new_builder()
- .column("product", DataTypes.STRING())
- .column("amount", DataTypes.INT())
+schema = Schema.new_builder() \
+ .column("product", DataTypes.STRING()) \
+ .column("amount", DataTypes.INT()) \
.build()
-sink_descriptor = TableDescriptor.for_connector("filesystem")
- .schema(schema)
+sink_descriptor = TableDescriptor.for_connector("filesystem") \
+ .schema(schema) \
.format(FormatDescriptor.for_format("csv")
.option("field-delimiter", ",")
- .build())
+ .build()) \
.build()
t_env.create_temporary_table("RubberOrders", sink_descriptor)
Review Comment:
This snippet still fails after the continuation fixes. The environment is
`table_env`, but this line calls `t_env.create_temporary_table(...)`, so it
raises a NameError. The Python descriptor also lacks the `path` option that the
Java tab sets, and the filesystem connector needs it. The zh page has the same
at [line
117](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/sql/reference/queries/overview.md#L117).
A few other Python snippets this PR touched still fail before any job runs:
- [zh
drop.md:125](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/sql/reference/ddl/drop.md#L125)
calls `table_env.listTables()`. PyFlink only has `list_tables()`, which the en
page uses.
- [zh
metrics.md:461](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/ops/metrics.md#L461)
calls `get_metric_group()`. The DataStream `RuntimeContext` has
`get_metrics_group()`, which the other snippets on the page use. The en Scope
snippet has the same name at [lines 377 and
382](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/ops/metrics.md#L377-L382).
-
[catalogs.md:423-429](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/sql/catalogs.md#L423-L429)
passes a string to `get_table` and `table_exists`, which take an `ObjectPath`.
`rename_table` also needs its `ignore_if_not_exists` argument. Same in
[zh](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/sql/catalogs.md#L418-L424).
-
[data_stream_api.md:1867](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/dev/table/data_stream_api.md#L1867)
uses `ChangelogMode.upsert()`, but the block does not import `ChangelogMode`.
Same in
[zh](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/dev/table/data_stream_api.md#L1869).
##########
docs/content/docs/ops/state/large_state_tuning.md:
##########
@@ -294,7 +294,7 @@ that the task-local state is an in-memory consisting of
heap objects, and not st
### Configuring task-local recovery
-Task-local recovery is *deactivated by default* and can be activated through
Flink's configuration with the key `state.backend.local-recovery` as specified
+Task-local recovery is *deactivated by default* and can be activated through
Flink's configuration with the key `execution.state-recovery.from-local` as
specified
in `CheckpointingOptions.LOCAL_RECOVERY`. The value for this setting can
either be *true* to enable or *false* (default) to disable local recovery.
Review Comment:
The key is right now, but the same sentence still says
`CheckpointingOptions.LOCAL_RECOVERY`. The option is
[`StateRecoveryOptions.LOCAL_RECOVERY`](https://github.com/apache/flink/blob/a21fa009bc5064f81525f501dea7da4759adbfc7/flink-core/src/main/java/org/apache/flink/configuration/StateRecoveryOptions.java#L109).
The zh page has the same at [line
234](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/ops/state/large_state_tuning.md#L234).
A few other lines sit next to fixes in this PR and keep the old problem:
- [zh
trace_reporters.md:48](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/deployment/trace_reporters.md#L48)
still names
`org.apache.flink.common.metrics.OpenTelemetryTraceReporterFactory`, which does
not exist. Line 52 and the en page use
`org.apache.flink.traces.otel.OpenTelemetryTraceReporterFactory`.
-
[data-types.md:1812-1813](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/sql/reference/data-types.md#L1812-L1813)
has two SELECTs without `;`. The other two blocks in this file got the fix.
Same in
[zh](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/sql/reference/data-types.md#L1804-L1805).
- [zh
udfs.md:1259](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/dev/table/functions/udfs.md#L1259)
only swapped the key. It still says the job fails when the number of failures
reaches `max-retries`. The en sentence now says the call was retried
`max-retries` times, which is one failure more. Could the zh clause count
retries too?
##########
docs/content.zh/docs/dev/table/functions/udfs.md:
##########
@@ -1779,8 +1782,8 @@ public static class WeightedAvgAccum {
public int count = 0;
}
-// The java class must have a public no-argument constructor and can be
founded in current java classloader.
-// Java 类必须有一个 public 的无参构造函数，并且可以在当前类加载器中加载到。
+# The java class must have a public no-argument constructor and can be founded
in current java classloader.
+# Java 类必须有一个 public 的无参构造函数，并且可以在当前类加载器中加载到。
Review Comment:
These two lines sit inside the `'''Java code: ...'''` string that runs from
line 1774 to line 1831. Python never parses them, so `//` was valid here and
right for the Java text. With `#` the embedded Java is no longer valid Java.
Could you revert this hunk?
##########
docs/content/docs/dev/datastream/fault-tolerance/state.md:
##########
@@ -704,70 +751,18 @@ it simply implies that the basic even-split
redistribution scheme will be used.
After initializing the container, we use the `isRestored()` method of the
context to check if we are
recovering after a failure. If this is `true`, *i.e.* we are recovering, the
restore logic is applied.
-As shown in the code of the modified `BufferingSink`, this `ListState`
recovered during state
-initialization is kept in a class variable for future use in
`snapshotState()`. There the `ListState` is cleared
-of all objects included by the previous checkpoint, and is then filled with
the new ones we want to checkpoint.
+Keep the `ListState` in a field for later use in `snapshotState()`. There,
update it with the objects
+you want to checkpoint, which replaces the objects included by the previous
checkpoint.
Review Comment:
After this change the page has no complete `CheckpointedFunction` example.
The Sink V2 writer is now the only full example, and [line
629](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/dev/datastream/fault-tolerance/state.md#L629)
says it does not implement that interface. These two paragraphs still describe
`isRestored()` and keeping the `ListState` for `snapshotState()`, but no
snippet shows either. Could we keep a small non-sink `CheckpointedFunction`
example with `snapshotState`, `initializeState` and the `isRestored()` branch,
and show the Sink V2 writer next to it? Dropping these two paragraphs would
work too. The same text is in
[state_v2.md](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/dev/datastream/fault-tolerance/state_v2.md#L690-L694)
and in the zh
[state.md](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/dev/datastrea
m/fault-tolerance/state.md#L632-L634) and
[state_v2.md](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/dev/datastream/fault-tolerance/state_v2.md#L689).
One more thing: [`flush(boolean
endOfInput)`](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/dev/datastream/fault-tolerance/state.md#L687)
is empty, so a bounded input drops the last buffered records. Should it send
the buffer there?
##########
docs/content.zh/docs/deployment/memory/network_mem_tuning.md:
##########
@@ -155,17 +150,16 @@ As a rule of thumb, we don't recommend thinking about
increasing the buffer size
- 内存使用高
- 大量的 checkpoint 数据量（针对非对齐的 checkpoints）
- 漫长的 checkpoint 时间（针对对齐的 checkpoints）
-- `execution.buffer-timeout` 较小时内存分配使用率会比较低，因为缓冲区还没被塞满数据就被发送下去了。
+- `execution.buffer-timeout.interval` 较小时内存分配使用率会比较低，因为缓冲区还没被塞满数据就被发送下去了。
### 选择缓冲区的数量
-缓冲区的数量是通过 `taskmanager.network.memory.buffers-per-channel` 和
`taskmanager.network.memory.floating-buffers-per-gate` 来配置的。
-
-为了最好的吞吐率，我们建议使用独占缓冲区和流动缓冲区的默认值(except you have one of [limit cases]({{< ref
"docs/deployment/memory/network_mem_tuning"
>}}#limitations))。如果缓冲数据量存在问题，更建议打开[缓冲消胀]({{< ref
"docs/deployment/memory/network_mem_tuning"
>}}#the-buffer-debloating-mechanism)。
+从 Flink 2.0 开始，缓冲区的数量不再可配置：Flink 为每个通道使用 2 个独占缓冲区，为每个 gate 使用 8 个流动缓冲区。
+如果缓冲数据量存在问题，建议打开[缓冲消胀]({{< ref "docs/deployment/memory/network_mem_tuning"
>}}#the-buffer-debloating-mechanism)。
Review Comment:
Nit: I think this anchor does not resolve. The zh heading at [line
41](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/deployment/memory/network_mem_tuning.md#L41)
is in Chinese, so its generated id is not `the-buffer-debloating-mechanism`.
It was like this before, but since the line is rewritten here, could it point
at the zh heading id? An explicit id on the heading would work too.
##########
docs/content/docs/concepts/sql-table-concepts/time_attributes.md:
##########
@@ -207,62 +207,34 @@ There are two ways of defining the time attribute when
converting a `DataStream`
In either case, the event time timestamp field will hold the value of the
`DataStream` event time timestamp.
+In Java, declare the event time attribute in the `Schema` that you pass to
`fromDataStream`: the `rowtime`
+metadata column holds the timestamp of each stream record as
`TIMESTAMP_LTZ(3)`, and `SOURCE_WATERMARK()` reuses
+the watermarks of the `DataStream`. The two options above use the
expression-based conversion, which is
+deprecated in Java. See [DataStream API Integration]({{< ref
"docs/dev/table/data_stream_api" >}}) for details.
Review Comment:
Nit: this paragraph correctly says the `rowtime` metadata column is
`TIMESTAMP_LTZ(3)`. [Line
203](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/concepts/sql-table-concepts/time_attributes.md#L203)
still says Flink always derives the rowtime attribute as TIMESTAMP WITHOUT
TIME ZONE in UTC. Could we scope that sentence to the expression-based
conversion? Same in [zh line
300](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/concepts/sql-table-concepts/time_attributes.md#L300).
##########
docs/content/docs/dev/table/data_stream_api.md:
##########
@@ -3410,44 +2595,6 @@ Table table = tableEnv.fromDataStream(stream, $("f1"));
// convert DataStream into Table with reordered and aliased fields "myString",
"myLong" (name-based)
Table table = tableEnv.fromDataStream(stream, $("f1").as("myString"),
$("f0").as("myLong"));
-```
-{{< /tab >}}
-{{< tab "Scala" >}}
-Flink supports Scala's built-in tuples.
-DataStreams of Scala's built-in tuples can be converted into tables.
-Fields can be renamed by providing names for all fields (mapping based on
position).
-If no field names are specified, the default field names are used.
-If the original field names (`_1`, `_2`, ... for Scala Tuples) are referenced,
the API assumes that the mapping is name-based instead of position-based.
-Name-based mapping allows for reordering fields and projection with alias
(`as`).
-
-```scala
-// get a TableEnvironment
-val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment"
section
-
-val stream: DataStream[(Long, String)] = ...
-
-// convert DataStream into Table with field names "myLong", "myString"
(position-based)
-val table: Table = tableEnv.fromDataStream(stream, $"myLong", $"myString")
-
-// convert DataStream into Table with reordered fields "_2", "_1" (name-based)
-val table: Table = tableEnv.fromDataStream(stream, $"_2", $"_1")
-
-// convert DataStream into Table with projected field "_2" (name-based)
-val table: Table = tableEnv.fromDataStream(stream, $"_2")
-
-// convert DataStream into Table with reordered and aliased fields "myString",
"myLong" (name-based)
-val table: Table = tableEnv.fromDataStream(stream, $"_2" as "myString", $"_1"
as "myLong")
-
-// define case class
-case class Person(name: String, age: Int)
-val streamCC: DataStream[Person] = ...
-
-// convert DataStream into Table with field names 'myName, 'myAge
(position-based)
-val table = tableEnv.fromDataStream(streamCC, $"myName", $"myAge")
-
-// convert DataStream into Table with reordered and aliased fields "myAge",
"myName" (name-based)
-val table: Table = tableEnv.fromDataStream(stream, $"age" as "myAge", $"name"
as "myName")
-
```
{{< /tab >}}
{{< tab "Python" >}}
Review Comment:
Nit: the Scala tab under this heading is gone, but the heading at [line
2571](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/dev/table/data_stream_api.md#L2571)
still says "Tuples (Scala, Java, Python) and Case Classes (Scala only)". The
POJO heading at [line
2629](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/dev/table/data_stream_api.md#L2629)
still says "Java and Scala". Could we rename both to match the tabs? The zh
page has them at [line
2560](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/dev/table/data_stream_api.md#L2560)
and [line
2626](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content.zh/docs/dev/table/data_stream_api.md#L2626).
##########
docs/content/docs/deployment/resource-providers/standalone/working_directory.md:
##########
@@ -54,21 +54,21 @@ If not explicitly configured, then it will be a random
value containing the host
Flink processes will use the working directory to store the following
artifacts:
* Blobs stored by the `BlobServer` and `BlobCache`
-* Local state if `state.backend.local-recovery` is enabled
+* Local state if `execution.state-recovery.from-local` is enabled
* RocksDB's working directory
## Local Recovery Across Process Restarts
The working directory can be used to enable local recovery across process
restarts ([FLIP-201](https://cwiki.apache.org/confluence/x/wJuqCw)).
This means that Flink does not have to recover state information from remote
storage.
-In order to use this feature, local recovery has to be enabled via
`state.backend.local-recovery`.
+In order to use this feature, local recovery has to be enabled via
`execution.state-recovery.from-local`.
Moreover, the TaskManager processes need to get a deterministic resource id
assigned via `taskmanager.resource-id`.
Last but not least, a failed TaskManager process needs to be restarted with
the same working directory.
```bash
process.working-dir: /path/to/working/dir/base
-state.backend.local-recovery: true
+execution.state-recovery.from-local: true
taskmanager.resource-id: TaskManager_1 # important: Change for every
TaskManager process
```
Review Comment:
Nit: this PR moves the failure enricher and listener snippets to a `yaml`
fence. This block is config.yaml content too and still uses `bash`. The same
goes for the untagged fences at
[s3.md:286](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/deployment/filesystems/s3.md#L286)
and [checkpointing_under_backpressure.md:97 and
:113](https://github.com/apache/flink/blob/823b71801f1c6cc5757963549059812736357e8c/docs/content/docs/ops/state/checkpointing_under_backpressure.md#L97-L113).
Could they use `yaml` too, with their zh copies?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
