text:
#ifndef M3BP_TEST_UTIL_PROCESSORS_BROADCAST_DUPLICATOR_HPP
#define M3BP_TEST_UTIL_PROCESSORS_BROADCAST_DUPLICATOR_HPP

#include <gtest/gtest.h>
#include <vector>
#include <algorithm>
#include "m3bp/processor_base.hpp"
#include "m3bp/task.hpp"
#include "util/generator_util.hpp"
#include "util/processors/test_processor_base.hpp"
#include "util/processors/input_deserializer.hpp"
#include "util/processors/output_builder.hpp"

namespace util {
namespace processors {

template <typename T>
class TestBroadcastDuplicator : public TestProcessorBase {

private:
	std::vector<int> m_ran_flags;

public:
	explicit TestBroadcastDuplicator(m3bp::size_type task_count)
		: TestProcessorBase(
			{ m3bp::InputPort("input0").movement(m3bp::Movement::BROADCAST) },
			{ m3bp::OutputPort("output0") })
		, m_ran_flags(task_count)
	{ }

	virtual void global_initialize(m3bp::Task &task) override {
		TestProcessorBase::global_initialize(task);
		task_count(m_ran_flags.size());
		std::fill(m_ran_flags.begin(), m_ran_flags.end(), false);
	}

	virtual void global_finalize(m3bp::Task &task) override {
		TestProcessorBase::global_finalize(task);
		for(const auto &flag : m_ran_flags){
			EXPECT_TRUE(flag);
		}
	}

	virtual void run(m3bp::Task &task) override {
		TestProcessorBase::run(task);

		const auto task_id = task.physical_task_id();
		EXPECT_FALSE(m_ran_flags[task_id]);
		m_ran_flags[task_id] = true;

		auto reader = task.input(0);
		const auto received = deserialize_value_only_buffer<T>(reader);
		auto writer = task.output(0);
		auto buffer = build_output_buffer(writer, received);
		writer.flush_buffer(std::move(buffer), received.size());
	}

};

}
}

#endif

redpajama_set_name: RedPajamaGithub


text:
package org.zouzias.swaggerjaxrswink.exception;

public class NotFoundException extends ApiException {
  private final int code;
  public NotFoundException (int code, String msg) {
    super(code, msg);
    this.code = code;
  }
}


text:
package org.apache.phoenix.transaction;

import java.io.Closeable;

public interface PhoenixTransactionClient extends Closeable {
}


text:
package io.kubernetes.client.openapi.models;

import io.kubernetes.client.fluent.Fluent;

/** Generated */
public interface V1CrossVersionObjectReferenceFluent<
        A extends V1CrossVersionObjectReferenceFluent<A>>
    extends Fluent<A> {
  public String getApiVersion();

  public A withApiVersion(String apiVersion);

  public Boolean hasApiVersion();

  public String getKind();

  public A withKind(String kind);

  public Boolean hasKind();

  public String getName();

  public A withName(String name);

  public Boolean hasName();
}


text:
package expr

import (
	"math"
	"strconv"
	"testing"

	"github.com/grafana/metrictank/api/models"
	"github.com/grafana/metrictank/schema"
	"github.com/grafana/metrictank/test"
)

var basic = []schema.Point{
	{Val: 0, Ts: 10},
	{Val: 10, Ts: 20},
	{Val: 20, Ts: 30},
}

var basicR = []schema.Point{
	{Val: 0, Ts: 10},
	{Val: float64(10) / 20, Ts: 20},
	{Val: 1, Ts: 30},
}

var nan = []schema.Point{
	{Val: 0, Ts: 10},
	{Val: math.NaN(), Ts: 20},
	{Val: 20, Ts: 30},
}

var nanR = []schema.Point{
	{Val: 0, Ts: 10},
	{Val: math.NaN() / 20, Ts: 20},
	{Val: 1, Ts: 30},
}

var minMaxSame = []schema.Point{
	{Val: 20, Ts: 10},
	{Val: 20, Ts: 20},
	{Val: 20, Ts: 30},
}

var minMaxSameR = []schema.Point{
	{Val: 0, Ts: 10},
	{Val: 0, Ts: 20},
	{Val: 0, Ts: 30},
}

var infinity = []schema.Point{
	{Val: 100, Ts: 10},
	{Val: 1000, Ts: 20},
	{Val: math.Inf(0), Ts: 30},
}

// infinity / infinity is undefined
var infinityR = []schema.Point{
	{Val: 0, Ts: 10},
	{Val: 0, Ts: 20},
	{Val: math.NaN(), Ts: 30},
}

func TestMinMaxBasic(t *testing.T) {
	testMinMax("NaN",
		[]models.Series{
			getSeries("targetHere", "queryPattHere", basic),
		},
		[]models.Series{
			getSeries("minMax(targetHere)", "minMax(queryPattHere)", basicR),
		},
		t)
}

func TestMinMaxNaN(t *testing.T) {
	testMinMax("NaN",
		[]models.Series{
			getSeries("targetHere", "queryPattHere", nan),
		},
		[]models.Series{
			getSeries("minMax(targetHere)", "minMax(queryPattHere)", nanR),
		},
		t)
}

func TestMinMaxSame(t *testing.T) {
	testMinMax("SameMinMax",
		[]models.Series{
			getSeries("targetHere", "queryPattHere", minMaxSame),
		},
		[]models.Series{
			getSeries("minMax(targetHere)", "minMax(queryPattHere)", minMaxSameR),
		},
		t)
}

func TestMinMaxInfinity(t *testing.T) {
	testMinMax("Infinity",
		[]models.Series{
			getSeries("targetHere", "queryPattHere", infinity),
		},
		[]models.Series{
			getSeries("minMax(targetHere)", "minMax(queryPattHere)", infinityR),
		},
		t)
}

func TestMinMaxZero(t *testing.T) {
	testMinMax("Zero",
		[]models.Series{
			getSeries("targetHere", "queryPattHere", []schema.Point{}),
		},
		[]models.Series{
			getSeries("minMax(targetHere)", "minMax(queryPattHere)", []schema.Point{}),
		},
		t)
}

func TestMinMaxMultiple(t *testing.T) {
	testMinMax("MultipleSeries",
		[]models.Series{
			getSeries("targetHere", "queryPattHere", basic),
			getSeries("targetHere2", "queryPattHere2", basic),
		}, []models.Series{
			getSeries("minMax(targetHere)", "minMax(queryPattHere)", basicR),
			getSeries("minMax(targetHere2)", "minMax(queryPattHere2)", basicR),
		},
		t)
}

func getNewMinMax(in []models.Series) *FuncMinMax {
	f := NewMinMax()
	ps := f.(*FuncMinMax)
	ps.in = NewMock(in)
	return ps
}

func testMinMax(name string, in []models.Series, out []models.Series, t *testing.T) {
	f := getNewMinMax(in)

	inputCopy := models.SeriesCopy(in) // to later verify that it is unchanged

	dataMap := initDataMap(in)

	got, err := f.Exec(dataMap)
	if err := equalOutput(out, got, nil, err); err != nil {
		t.Fatalf("Case %s: %s", name, err)
	}

	t.Run("DidNotModifyInput", func(t *testing.T) {
		if err := equalOutput(inputCopy, in, nil, nil); err != nil {
			t.Fatalf("Case %s: Input was modified, err = %s", name, err)
		}
	})

	t.Run("DoesNotDoubleReturnPoints", func(t *testing.T) {
		if err := dataMap.CheckForOverlappingPoints(); err != nil {
			t.Fatalf("Case %s: Point slices in datamap overlap, err = %s", name, err)
		}
	})
}

func BenchmarkMinMax10k_1NoNulls(b *testing.B) {
	benchmarkMinMax(b, 1, test.RandFloats10k, test.RandFloats10k)
}
func BenchmarkMinMax10k_10NoNulls(b *testing.B) {
	benchmarkMinMax(b, 10, test.RandFloats10k, test.RandFloats10k)
}
func BenchmarkMinMax10k_100NoNulls(b *testing.B) {
	benchmarkMinMax(b, 100, test.RandFloats10k, test.RandFloats10k)
}
func BenchmarkMinMax10k_1000NoNulls(b *testing.B) {
	benchmarkMinMax(b, 1000, test.RandFloats10k, test.RandFloats10k)
}
func BenchmarkMinMax10k_1SomeSeriesHalfNulls(b *testing.B) {
	benchmarkMinMax(b, 1, test.RandFloats10k, test.RandFloatsWithNulls10k)
}
func BenchmarkMinMax10k_10SomeSeriesHalfNulls(b *testing.B) {
	benchmarkMinMax(b, 10, test.RandFloats10k, test.RandFloatsWithNulls10k)
}
func BenchmarkMinMax10k_100SomeSeriesHalfNulls(b *testing.B) {
	benchmarkMinMax(b, 100, test.RandFloats10k, test.RandFloatsWithNulls10k)
}
func BenchmarkMinMax10k_1000SomeSeriesHalfNulls(b *testing.B) {
	benchmarkMinMax(b, 1000, test.RandFloats10k, test.RandFloatsWithNulls10k)
}
func BenchmarkMinMax10k_1AllSeriesHalfNulls(b *testing.B) {
	benchmarkMinMax(b, 1, test.RandFloatsWithNulls10k, test.RandFloatsWithNulls10k)
}
func BenchmarkMinMax10k_10AllSeriesHalfNulls(b *testing.B) {
	benchmarkMinMax(b, 10, test.RandFloatsWithNulls10k, test.RandFloatsWithNulls10k)
}
func BenchmarkMinMax10k_100AllSeriesHalfNulls(b *testing.B) {
	benchmarkMinMax(b, 100, test.RandFloatsWithNulls10k, test.RandFloatsWithNulls10k)
}
func BenchmarkMinMax10k_1000AllSeriesHalfNulls(b *testing.B) {
	benchmarkMinMax(b, 1000, test.RandFloatsWithNulls10k, test.RandFloatsWithNulls10k)
}

func benchmarkMinMax(b *testing.B, numSeries int, fn0, fn1 func() []schema.Point) {
	var input []models.Series
	for i := 0; i < numSeries; i++ {
		series := models.Series{
			QueryPatt: strconv.Itoa(i),
		}
		if i%2 == 0 {
			series.Datapoints = fn0()
		} else {
			series.Datapoints = fn1()
		}
		input = append(input, series)
	}
	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		f := NewMinMax()
		f.(*FuncMinMax).in = NewMock(input)
		got, err := f.Exec(make(map[Req][]models.Series))
		if err != nil {
			b.Fatalf("%s", err)
		}
		results = got
	}
}


text:
using System.Reflection;
using System.Resources;
using System.Runtime.CompilerServices;

[assembly: AssemblyTitle("NGauge.Runner")]
[assembly: AssemblyDescription("Executes Gauge-to-NCrunch bridged-steps generated by NGauge.")]
[assembly: AssemblyConfiguration("")]
[assembly: AssemblyTrademark("")]
[assembly: AssemblyCulture("")]
[assembly: NeutralResourcesLanguage("en")]
[assembly: InternalsVisibleTo("NGauge.Runner.Tests, PublicKey=00240000048000009400000006020000002400005253413100040000010001007f0802aaf454c2156ca274cd785e6b87e2e8ec2c8b7eaf3e7d3612a63b6f7fe76cf5b0fb018e2e08d65eaf9b5c083afb4cd33aa04eaab404d1a0de8442fa0ff4fab237f54d0bea7fbb41a4f5160edbd71576bb44a7b9ed90565425c876c8a498887580cc516b3360aaac33e0508131dcd7cf8bd053a7c84d03626809240b28d6")]


text:
+++
title = "Dragonfly Cache Design"
tags = ["dragonfly", "redis"]
categories = []
description = ""
menu = ""
banner = ""
images = []
date = 2022-06-23T12:00:00+03:00
draft = false
+++

I talked in my [previous post]({{< ref "simple_is_beatiful.md" >}}) about Redis eviction policies.
In this post, I would like to describe the design behind Dragonfly cache.
<!--more-->

If you have not heard about [Dragonfly](https://github.com/dragonflydb/dragonfly) - please check it out. It uses - what I hope - novel and interesting ideas backed up by the  research from recent years [[1]](https://doi.org/10.1007/s00778-014-0377-7)
and [[2]](https://arxiv.org/abs/2003.07302). It's meant to fix many problems that exist with Redis today. I have been working on Dragonfly for the last 7 months and it has been one of the more interesting and challenging projects I've ever done!

Anyway, back to cache design. We'll start with a short overview of what LRU cache is and its shortcomings in general, and then analyze Redis implementation specifically.

## LRU
The least recently used (LRU) cache policy evicts items, as the name implies, that were least recently used. It works this way because a cache algorithm strives to optimize the hit ratio, or the probability that its items will be accessed in the future.

If a cache is full, it needs to vacate items to make room for new additions. The cache frees space for new additions by removing the least valuable items, **assuming** that the *least recently used* item is also the *least valuable*.

The assumption is reasonable, but unfortunately, this algorithm behaves poorly if the assumption above does not hold. Consider, for example, an access pattern with [Long tail distribution](https://en.wikipedia.org/wiki/Long_tail). Here, the y-axis represents the normalized access frequency of the items in cache and the x-axis represents the items ordered from highest frequency to lowest.

{{< figure src="https://upload.wikimedia.org/wikipedia/commons/thumb/8/8a/Long_tail.svg/1920px-Long_tail.svg.png" alt="long tail distribution" width="350px">}}

In this case, lots of newly added "yellow" items with low access frequency could push out rare but valuable "green" items responsible for the vast majority of hits. As a result, the LRU policy may sweep its contents together with valuable "green" entities due to traffic fluctuations.

{{< figure src="/img/lru.svg">}}

### LRU implementation efficiency
LRU is a simple algorithm that can be implemented efficiently.
Indeed, it maintains all the items in a double-linked list. When an item is accessed, LRU moves it to the head of the list. To vacate LRU items, it pops from the tail of the list. See the diagram above. All operations are done in `O(1)`, and the memory overhead per item is 2 pointers, i.e. 16 bytes on 64bit architecture.

### LRU in Redis
Redis implements a few eviction policy heuristics. Some of them are described as "approximated LRU". Why approximated? Because Redis does not maintain an exact global order among its items like in classic LRU. Instead, it stores the last access timestamp in each entry.

When it needs to evict an item, Redis performs random sampling of the entire keyspace and selects K candidates. Then it choosed the item with the least recently used timestamp among those K candidates and vacates it. This way, Redis saves 16 bytes per entry needed for ordering items in one global order. This heuristic is a very rough approximation of an LRU.Redis maintainers [have recently discussed](https://github.com/redis/redis/issues/8947) the possibility of adding an additional heuristic to Redis by implementing a classic LRU policy with global order but eventually decided against it.

## Dragonfly Cache
Dragonfly implements cache that:
  * is resistant to fluctuations of recent traffic, unlike LRU.
  * Does not require random sampling or other approximations like in Redis.
  * Has **zero** memory overhead per item.
  * Has very small `O(1)` run-time overhead.

It's a novel approach for cache design has not been suggested before in academic research.

Dragonfly Cache (dash cache) is based on another famous cache policy from [1994 paper](https://api.semanticscholar.org/CorpusID:6259428) - "2Q: A Low Overhead High Performance Buffer Management Replacement Algorithm".

2Q addresses the issues with LRU by introducing two independent buffers. Instead of considering just recency as a factor, 2Q also considers access frequency for each item. It first admits recent items into a so-called *probationary buffer* (see below). This buffer holds only a little part of the cache space, say less than 10%. All newly added items compete with each other inside this buffer.

{{< figure src="/img/2q-cache.svg">}}

Only if a probationary item was accessed at least once, will it be proven as *worthy*, and upgraded to the protected buffer. By doing so, it evicts the least recently used item from the protected buffer back to probationary buffer. You can read [this post](https://arpitbhayani.me/blogs/2q-cache) for more details.

2Q improves LRU by acknowledging that just because a new item was added to the cache - does not mean it's useful. 2Q requires an to have been accessed at least once before to be considered as a high quality item. In this way, 2Q cache has been proven to be more robust and to achieve a higher hit rate than LRU policy.

### Dashtable in 60 seconds
For a deep dive on Dashtable's use in Dragonfly, I recommend you reading this [post](https://github.com/dragonflydb/dragonfly/blob/main/docs/dashtable.md) or [the original paper](https://arxiv.org/abs/2003.07302).

For our purposes today, we need to know the following facts about Dashtable in Dragonfly implementation:

1. It's comprised of segments of constant size. Each segment holds 56 regular buckets with multiple slots. Each slot has space for a single item.
2. Dashtable's routing algorithm uses item's hash value to compute its segment id. In addition, it predefines 2 out of 56 buckets where the item can reside within the segment. The item can reside in any free slot in those two buckets.
3. In addition to regular buckets, a Dashtable segment manages 4 stash buckets that may gather overflow items that do not have space in their assigned buckets. The routing algorithm never assigns stash buckets directly. Instead, only if its 2 home buckets are full, the item is allowed to reside in any of 4 stash buckets. This greatly increases segment's utilization.
4. Once a segment becomes full, and there is no free place for a new item in its home buckets nor in stash buckets, the Dashtable grows by adding a new segment and splitting the contents of the full segment roughly in half.

{{< figure src="/img/segment.svg">}}

The point at which a segment becomes full is a convenient time to add any type of eviction policy to Dashtable, because this is when Dashtable grows. Moreover, in order to prevent growth of Dashtable, we can only evict items from a full segment. This constitutes a very precise eviction framework that operates in `O(1)` run-time complexity.

### 2Q implementation

Dragonfly expands on the ideas above. A naive solution would be to divide hashtable entries into two buffers: a probationary buffer with FIFO ordering, and the protected buffer employing LRU linked-list. That would work but it would require using additional metadata and waste precious memory.

Instead, Dragonfly leverages the unique design of [Dashtable](https://github.com/dragonflydb/dragonfly/blob/main/doc/dashtable.md) and uses its weak ordering characteristics for its advantage.

To explain how 2Q works with Dashtable, we need to explain how we define probationary and protected buffers there, how we promote a probationary item into protected buffer and how we evict items from the cache.

{{< figure src="/img/dash-cache.svg">}}

We overlaid the following semantics on the original Dashtable:
1. Slots within a bucket now have rank or precedence. A slot on the left has the highest rank `(0)`, and the last slot on the right has the lowest rank `(9)`.
2. Stash buckets within a segment serve as a probationary buffer. When a new item is added to the full segment, it's been added into a stash bucket at slot 0. All other items in the bucket are shifted right, and the last item in the bucket is evicted. This way the bucket serves as a FIFO queue for probationary items.
3. Every cache hit "promotes" its item:
   * if the item was in a stash bucket, it's moved immediately into its home bucket to the last slot.
   * if it was in a home bucket at slot `i`, it's swapped out with an item at slot `i-1`.
   * an item at slot `0` stays in the same place.
4. When a probationary item is promoted to the protected bucket, it's moved to the last slot there. The item that was there before is demoted back into the probationary bucket.

Basically, Dash-Cache eviction policy is comprised of an "eviction" step described by (2) and the positive reinforcement step described by (3).

That's it. No additional metadata is needed. High quality items will come to reside at high rank slots in their home buckets very quickly, while newly added items compete with each other within stash/probationary buckets. In our implementation each bucket has 14 slots, meaning that each probationary item can be shifted 14 times before being evicted from the cache, unless it proves its usefulness and is promoted. Each segment has 56 regular buckets and 4 stash buckets; therefore we have `6.7%` of total space allocated for a probationary buffer. It's enough to catch high quality items before they are evicted.

I hope you enjoyed reading how Dragonfly cache design utilizes seemingly unrelated concepts together to its advantage.

*I want to thank [Ben Manes](https://twitter.com/benmanes), the author of wonderful [caffeine package](https://github.com/ben-manes/caffeine) for early comments and the guidance on how to use caffeine simulator in order to compare Dragonfly Cache to other caches.*


text:
/**
 * Lit-it  : Add character to your JSDocs
 * Author  : Mohseen Mukaddam [mohseenmukaddam6@gmail.com]
 * License : Apache 2.0
 */
import * as vscode from 'vscode';
import {
    Range,
    Position,
    TextEdit,
    Disposable,
    TextEditor,
    WorkspaceEdit,
    ExtensionContext
} from 'vscode';
let count: number = 0;

// https://code.visualstudio.com/Docs/extensionAPI/vscode-api#_workspace
// subscribe to onDidChangeTextDocument and add move selection

/**
 * @function activate
 * @param  {type} context {description}
 * @return {type} {description}
 */
export function activate ( context: ExtensionContext ): void {

    // Use the console to output diagnostic information (console.log) and errors (console.error)
    // This line of code will only be executed once when your extension is activated
    console.log( 'Congratulations, your extension "Lit - it" is now active!' );

    // The command has been defined in the package.json file
    // Now provide the implementation of the command with  registerCommand
    // The commandId parameter must match the command field in package.json
    const disposable: Disposable = vscode.commands.registerCommand( 'extension.litIt', () => {

        // doc reference
        // https://code.visualstudio.com/Docs/extensionAPI/vscode-api#Range
        // https://code.visualstudio.com/Docs/extensionAPI/vscode-api
        const editor: TextEditor = vscode.window.activeTextEditor;

        if ( editor.selection.isEmpty ) {
            const position: Position = editor.selection.active;
            console.log( position );
            // test case if line is 1;
            // get below line
            const offset: number = 2;
            const requiredRange: Range = new Range( ( position.line + 1 ), 0, ( position.line + offset ), 0 );

            const textOfInterest: string = editor.document.getText( requiredRange );
            console.log( textOfInterest );

            // generate docString
            const docString: string = functionDocString( textOfInterest );

            // insert this at current position
            const insertionText: TextEdit = new TextEdit( new Range( ( position.line ), 0, ( position.line + 1 ), 0 ), docString );

            const workSpaceEdit: WorkspaceEdit = new vscode.WorkspaceEdit();
            workSpaceEdit.set( editor.document.uri, [ insertionText ] );

            // apply edit
            vscode.workspace.applyEdit( workSpaceEdit );
        } else { vscode.window.showInformationMessage( 'Lit-it does not work with selection' ); }
        const dispose: Disposable = vscode.workspace.onDidChangeTextDocument( tabEventListner );
        // probably use promises to dispose the event listener

    });

    context.subscriptions.push( disposable );
}

// this method is called when your extension is deactivated
export function deactivate (): boolean {
    return true;
}

// file signature
export function checkSignature ( signature: string ): string {
    if ( signature.includes( 'function' ) ) {
        return 'FUNCTION';
    }
    return 'ES6';
}
/**
 * @author Mohseen Mukaddam <mohseenmukaddam6@gmail.com>
 */

export function functionDocString ( signature: string ): string {
    const indentNum: number = signature.search(/\S/);
    let indent: string = '';
    for (let i: number = 0; i < indentNum; i++) {
        indent += ' ';
    }
    let template: string = ``;
    template += `${indent}/**\n`;
    template += `${indent} * @function `;

    const name: string = extractFunctionName( signature );
    template += `${(name === '') ? `{function name}\n` : `${name}\n`}`;

    const parameters: string[] = extractParameters( signature );

    if ( parameters.length === 0 ) {
        template += `${indent} * @return {type} {description}\n${indent} */\n`;
        return template;
    } else {
        const prettyParams: string[] = prettyParameters( parameters );
        const parameterStrings: string[] = prettyParams.map( (param: string) => `${indent} * @param  {type} ${param} {description}\n` );
        let parameterString: string = parameterStrings.reduce( ( acc: string, curr: string ) => acc.concat( curr ) ).toString();
        parameterString += `${indent} * @return {type} {description}\n${indent} */\n`;
        return( template.concat( parameterString ) );
    }
}

/**
 * @function extractParameters
 * @param  {String} signature The function signature
 * @return {List} List of parameters
 */
export function extractParameters ( signature: string ): string[] {
    const possibleParameters: string = signature.slice( signature.indexOf( '(' ) + 1, signature.indexOf( ')' ) ).trim();
    const filteredArguments: string[] = ( possibleParameters === '' ) ? [] : possibleParameters.split( ',' ).map( ( str: string ) => str.trim() );
    const filteredArgumentsReturned: string[] = filteredArguments.map( mapES6Destructuring );
    return filteredArgumentsReturned;
}
function mapES6Destructuring ( arg: string ): string {
    let arrayOfStrings: string = arg;
    if ( arg.includes( '{' ) ) {
        arrayOfStrings = cleanFirstBracketCase( arg );
    }
    if ( arg.includes( '}' ) ) {
        arrayOfStrings = cleanLastBracketCase( arg );
    }
    if ( arg.includes( '=' ) ) {
        arrayOfStrings = cleanEqualCase( arg );
    }
    return arrayOfStrings;
}
function cleanEqualCase ( arg: string ): string {
    if ( arg.includes( '=' ) ) {
        return arg.split( '=' )[ 0 ].trim();
    }
    return null;
}
function cleanFirstBracketCase ( arg: string ): string {
    if ( arg.includes( '{' ) ) {
        return arg.split( '{' )[ 1 ].trim();
    }
    return null;
}
function cleanLastBracketCase ( arg: string ): string {
    if ( arg.includes( '}' ) ) {
        return arg.split( '}' )[ 0 ].trim();
    }
    return null;
}
export function addPadding ( str: string, max: number ): string  {
    if ( str.length < max ) {
        let i: number = str.length;
        for ( ; i < max; i++ ) {
            str += ' ';
        }
        return str;
    }
    return str;
}

export function prettyParameters ( listOfParameters: string[] ): string[] {
    const lengths: number[] = listOfParameters.map( (str: string) => str.length );
    const max: any = Math.max.apply( null, lengths );

    const mappingFunction: (x: string) => string = ( x: string ) => addPadding( x, max );
    return listOfParameters.map( mappingFunction );
}

export function extractFunctionName ( signature: string ): string {
    // check if contains function -> else check var. let, const
    // if yes check name in front -> else check var, let, const
    if ( signature.includes( 'function' ) ) {
        const offset: number = 8;
        const currentPosition: number = signature.indexOf( 'function' ) + offset;
        const endSlicePosition: number = signature.indexOf( '(' );
        const name: string = signature.slice( currentPosition, endSlicePosition ).trim();
        return ( name !== '' ) ? name : extractAlternateFunctionName( signature );
    }
    return extractAlternateFunctionName( signature );
}

export function extractAlternateFunctionName ( signature: string ): string {
    if ( signature.includes( 'var' ) || signature.includes( 'let' ) || signature.includes( 'const' ) ) {
        if ( signature.includes( '=' ) ) {
            interface IMapIndex {
                var: number;
                const: number;
                let: number;
                [key: string]: number;
            }
            const mapIndex: IMapIndex = {
                var : signature.indexOf( 'var' ),
                const : signature.indexOf( 'const' ),
                let : signature.indexOf( 'let' ),
            };
            const keys: string[] = Object.keys( mapIndex ).filter( (x: string) => mapIndex[x] !== -1 );
            const startingIndex: number = mapIndex[ keys[0] ] + keys[0].length;

            return signature.slice( startingIndex, signature.indexOf( '=' ) ).trim();
        }

    }
    return '';
}

function tabEventListner (): void {
    count++;
}


text:
DocumentationEditor::Engine.routes.draw do
  constraints format: 'html' do
    scope path: '/admin' do
      get '/', as: :admin, :controller => 'pages', :action => 'index'
      post '/', :controller => 'pages', :action => 'create'

      post '/images', as: :upload_image, :controller => 'pages', :action => 'upload_image'

      get '/export', as: :export, :controller => 'pages', :action => 'export'
      post '/import', as: :import, :controller => 'pages', :action => 'import'

      get '/:id', :controller => 'pages', :action => 'source'
      get '/:id/edit', as: :edit_page, :controller => 'pages', :action => 'edit'
      put '/:id', as: :update, :controller => 'pages', :action => 'update'
      post '/:id', :controller => 'pages', :action => 'commit'
      post '/:id/thumbnail', as: :upload_thumbnail, :controller => 'pages', :action => 'upload_thumbnail'
      delete '/:id', as: :delete_page, :controller => 'pages', :action => 'destroy'
      get '/:id/preview', as: :preview_page, :controller => 'pages', :action => 'preview'
      get '/:prev/:cur/diff', :controller => 'pages', :action => 'diff'
      get '/:id/history', as: :history, :controller => 'pages', :action => 'history'
      get '/:id/versions', :controller => 'pages', :action => 'versions'
    end

    get '/:slug/sections/:section', as: :section, :controller => 'pages', :action => 'show', constraints: { slug: /.*/ }
    get '/:slug', as: :page, :controller => 'pages', :action => 'show', constraints: { slug: /.*/ }
  end
end


text:
package main

import (
	"os"
	"fmt"

	"github.com/amsokol/openshift-golang-template/example-golang-dep/pkg/cmd/cli"
	"github.com/amsokol/openshift-golang-template/example-golang-dep/pkg/cmd/server"
)

func main() {
	cli.EchoArgs()
	if err := server.Start(cli.GetPort()); err != nil {
		os.Exit(1)
	}
}


text:
function v = getval(adv)
% GETDER(ADV) returns the value of the advar ADV

% Copyright (c) 2013, Asgeir Birkisson and Tobin A. Driscoll.

v = adv.val;
end


text:
<?php

namespace Playbook;

use GuzzleHttp\Client as GuzzleClient;
use GuzzleHttp\Exception\ClientException;
use GuzzleHttp\Exception\ServerException;
use Playbook\Exception\ApplicantNotFoundException;
use Playbook\Exception\InvalidApplicantException;

class Client
{

    protected $client;
    protected $user;
    protected $api_key;
    protected $base_uri = "https://www.playbookhr.com";

    public function __construct($user, $api_key)
    {
        $this->user = $user;
        $this->api_key = $api_key;

        $this->client = new GuzzleClient([
            "base_uri" => $this->base_uri,
        ]);

    }

    /**
     * searches playbook for a user matching the applicant object criteria
     *
     * @param Applicant $applicant
     *
     * @return mixed
     * @throws ApplicantNotFoundException
     * @throws \Exception
     */
    public function searchApplicants(Applicant $applicant)
    {

        if (!$applicant->email) {
            throw new InvalidApplicantException("Applicant is missing a required field");
        }

        try {
            $response = $this->client->request("GET", '/api/v2/search', [
                "auth" => [$this->user, $this->api_key],
                "query" => $applicant->toArray(),
            ]);
        } catch (ClientException $e) {
            throw new \Exception($e->getMessage());
        } catch (ServerException $e) {
            throw new \Exception("ERROR: " . $e->getMessage());
        }

        $body = $response->getBody();
        $json = json_decode((string)$body, true);

        if ($json['found'] === false) {
            throw new ApplicantNotFoundException("Applicant was not found");
        }

        return $applicant->getNewInstance(array_filter($json['applicant']), $this);

    }

    /**
     * adds a applicant to playbook according to the applicant object values
     *
     * @param Applicant $applicant
     *
     * @return mixed
     * @throws \Exception
     */
    public function addApplicant(Applicant $applicant)
    {

        $this->validateApplicant($applicant);

        try {
            $response = $this->client->request("POST", '/api/v2/applicants', [
                "auth" => [$this->user, $this->api_key],
                "form_params" => $applicant->toArray(),
            ]);
        } catch (ClientException $e) {
            throw new \Exception($e->getMessage());
        } catch (ServerException $e) {
            throw new \Exception("ERROR: " . $e->getMessage());
        }

        $body = $response->getBody();
        $json = json_decode((string)$body, true);
        $applicant->assignProps($json);

        return $applicant->getNewInstance($applicant, $this);
    }

    /**
     * updates an applicant in playbook. Must pass validations.
     *
     * @param Applicant $applicant
     *
     * @return mixed
     * @throws \Exception
     */
    public function updateApplicant(Applicant $applicant)
    {

        $this->validateApplicant($applicant);

        try {
            $response = $this->client->request("PUT", '/api/v2/applicants', [
                "auth" => [$this->user, $this->api_key],
                "form_params" => $applicant->toArray(),
            ]);
        } catch (ClientException $e) {
            throw new \Exception($e->getMessage());
        } catch (ServerException $e) {
            throw new \Exception("ERROR: " . $e->getMessage());
        }

        $body = $response->getBody();
        $json = json_decode((string)$body, true);
        $applicant->assignProps($json);

        return $applicant->getNewInstance($applicant, $this);
    }

    /**
     * check that our applicant is valid or throw our exception
     *
     * @param Applicant $applicant
     */
    private function validateApplicant(Applicant $applicant)
    {
        if (!$applicant->isValid()) {
            throw new InvalidApplicantException("Applicant is missing a required field");
        }
    }

}


text:
title: "概念与通用 API"
nav-parent_id: tableapi
nav-pos: 0
---

The Table API and SQL are integrated in a joint API. The central concept of this API is a `Table` which serves as input and output of queries. This document shows the common structure of programs with Table API and SQL queries, how to register a `Table`, how to query a `Table`, and how to emit a `Table`.

* This will be replaced by the TOC
{:toc}

Main Differences Between the Two Planners
-----------------------------------------

1. Blink treats batch jobs as a special case of streaming. As such, the conversion between Table and DataSet is not supported, and batch jobs will not be translated into `DateSet` programs but translated into `DataStream` programs, the same as the streaming jobs.
2. The Blink planner does not support `BatchTableSource`, uses bounded `StreamTableSource` instead of it.
3. The Blink planner only support the brand new `Catalog` and does not support `ExternalCatalog` which is deprecated.
4. The implementations of `FilterableTableSource` for the old planner and the Blink planner are incompatible. The old planner will push down `PlannerExpression`s into `FilterableTableSource`, while the Blink planner will push down `Expression`s.
5. String based key-value config options (Please see the documentation about [Configuration]({{ site.baseurl }}/dev/table/config.html) for details) are only used for the Blink planner.
6. The implementation(`CalciteConfig`) of `PlannerConfig` in two planners is different.
7. The Blink planner will optimize multiple-sinks into one DAG (supported only on `TableEnvironment`, not on `StreamTableEnvironment`). The old planner will always optimize each sink into a new DAG, where all DAGs are independent of each other.
8. The old planner does not support catalog statistics now, while the Blink planner does.

Structure of Table API and SQL Programs
---------------------------------------

All Table API and SQL programs for batch and streaming follow the same pattern. The following code example shows the common structure of Table API and SQL programs.

{% highlight java %}

// create a TableEnvironment for specific planner batch or streaming
TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

// register a Table
tableEnv.registerTable("table1", ...)            // or
tableEnv.registerTableSource("table2", ...);
// register an output Table
tableEnv.registerTableSink("outputTable", ...);

// create a Table from a Table API query
Table tapiResult = tableEnv.scan("table1").select(...);
// create a Table from a SQL query
Table sqlResult  = tableEnv.sqlQuery("SELECT ... FROM table2 ... ");

// emit a Table API result Table to a TableSink, same for SQL result
tapiResult.insertInto("outputTable");

// execute
tableEnv.execute("java_job");

{% endhighlight %}

{% highlight scala %}

// create a TableEnvironment for specific planner batch or streaming
val tableEnv = ... // see "Create a TableEnvironment" section

// register a Table
tableEnv.registerTable("table1", ...)           // or
tableEnv.registerTableSource("table2", ...)
// register an output Table
tableEnv.registerTableSink("outputTable", ...);

// create a Table from a Table API query
val tapiResult = tableEnv.scan("table1").select(...)
// create a Table from a SQL query
val sqlResult  = tableEnv.sqlQuery("SELECT ... FROM table2 ...")

// emit a Table API result Table to a TableSink, same for SQL result
tapiResult.insertInto("outputTable")

// execute
tableEnv.execute("scala_job")

{% endhighlight %}

{% highlight python %}

# create a TableEnvironment for specific planner batch or streaming
table_env = ... # see "Create a TableEnvironment" section

# register a Table
table_env.register_table("table1", ...)           # or
table_env.register_table_source("table2", ...)

# register an output Table
table_env.register_table_sink("outputTable", ...);

# create a Table from a Table API query
tapi_result = table_env.scan("table1").select(...)
# create a Table from a SQL query
sql_result  = table_env.sql_query("SELECT ... FROM table2 ...")

# emit a Table API result Table to a TableSink, same for SQL result
tapi_result.insert_into("outputTable")

# execute
table_env.execute("python_job")

{% endhighlight %}

**Note:** Table API and SQL queries can be easily integrated with and embedded into DataStream or DataSet programs. Have a look at the [Integration with DataStream and DataSet API](#integration-with-datastream-and-dataset-api) section to learn how DataStreams and DataSets can be converted into Tables and vice versa.

{% top %}

Create a TableEnvironment
-------------------------

The `TableEnvironment` is a central concept of the Table API and SQL integration. It is responsible for:

* Registering a `Table` in the internal catalog
* Registering an external catalog
* Executing SQL queries
* Registering a user-defined (scalar, table, or aggregation) function
* Converting a `DataStream` or `DataSet` into a `Table`
* Holding a reference to an `ExecutionEnvironment` or `StreamExecutionEnvironment`

A `Table` is always bound to a specific `TableEnvironment`. It is not possible to combine tables of different TableEnvironments in the same query, e.g., to join or union them.

A `TableEnvironment` is created by calling the static `BatchTableEnvironment.create()` or `StreamTableEnvironment.create()` method with a `StreamExecutionEnvironment` or an `ExecutionEnvironment` and an optional `TableConfig`. The `TableConfig` can be used to configure the `TableEnvironment` or to customize the query optimization and translation process (see [Query Optimization](#query-optimization)).

Make sure to choose the specific planner `BatchTableEnvironment`/`StreamTableEnvironment` that matches your programming language.

If both planner jars are on the classpath (the default behavior), you should explicitly set which planner to use in the current program.

{% highlight java %}

// **********************
// FLINK STREAMING QUERY
// **********************
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.java.StreamTableEnvironment;

EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build();
StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings);
// or TableEnvironment fsTableEnv = TableEnvironment.create(fsSettings);

// ******************
// FLINK BATCH QUERY
// ******************
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.table.api.java.BatchTableEnvironment;

ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment fbTableEnv = BatchTableEnvironment.create(fbEnv);

// **********************
// BLINK STREAMING QUERY
// **********************
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.java.StreamTableEnvironment;

StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);
// or TableEnvironment bsTableEnv = TableEnvironment.create(bsSettings);

// ******************
// BLINK BATCH QUERY
// ******************
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

EnvironmentSettings bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build();
TableEnvironment bbTableEnv = TableEnvironment.create(bbSettings);

{% endhighlight %}

{% highlight scala %}

// **********************
// FLINK STREAMING QUERY
// **********************
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.scala.StreamTableEnvironment

val fsSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
val fsEnv = StreamExecutionEnvironment.getExecutionEnvironment
val fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings)
// or val fsTableEnv = TableEnvironment.create(fsSettings)

// ******************
// FLINK BATCH QUERY
// ******************
import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.table.api.scala.BatchTableEnvironment

val fbEnv = ExecutionEnvironment.getExecutionEnvironment
val fbTableEnv = BatchTableEnvironment.create(fbEnv)

// **********************
// BLINK STREAMING QUERY
// **********************
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.EnvironmentSettings
import org.apache.flink.table.api.scala.StreamTableEnvironment

val bsEnv = StreamExecutionEnvironment.getExecutionEnvironment
val bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build()
val bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings)
// or val bsTableEnv = TableEnvironment.create(bsSettings)

// ******************
// BLINK BATCH QUERY
// ******************
import org.apache.flink.table.api.{EnvironmentSettings, TableEnvironment}

val bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inBatchMode().build()
val bbTableEnv = TableEnvironment.create(bbSettings)

{% endhighlight %}

{% highlight python %}

# **********************
# FLINK STREAMING QUERY
# **********************
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings

f_s_env = StreamExecutionEnvironment.get_execution_environment()
f_s_settings = EnvironmentSettings.new_instance().use_old_planner().in_streaming_mode().build()
f_s_t_env = StreamTableEnvironment.create(f_s_env, environment_settings=f_s_settings)

# ******************
# FLINK BATCH QUERY
# ******************
from pyflink.dataset import ExecutionEnvironment
from pyflink.table import BatchTableEnvironment

f_b_env = ExecutionEnvironment.get_execution_environment()
f_b_t_env = BatchTableEnvironment.create(f_b_env, table_config)

# **********************
# BLINK STREAMING QUERY
# **********************
from pyflink.datastream import StreamExecutionEnvironment
from pyflink.table import StreamTableEnvironment, EnvironmentSettings

b_s_env = StreamExecutionEnvironment.get_execution_environment()
b_s_settings = EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
b_s_t_env = StreamTableEnvironment.create(b_s_env, environment_settings=b_s_settings)

# ******************
# BLINK BATCH QUERY
# ******************
from pyflink.table import EnvironmentSettings, BatchTableEnvironment

b_b_settings = EnvironmentSettings.new_instance().use_blink_planner().in_batch_mode().build()
b_b_t_env = BatchTableEnvironment.create(environment_settings=b_b_settings)

{% endhighlight %}

**Note:** If there is only one planner jar in `/lib` directory, you can use `useAnyPlanner` (`use_any_planner` for python) to create specific `EnvironmentSettings`.

{% top %}

Register Tables in the Catalog
-------------------------------

A `TableEnvironment` maintains a catalog of tables which are registered by name. There are two types of tables, *input tables* and *output tables*. Input tables can be referenced in Table API and SQL queries and provide input data. Output tables can be used to emit the result of a Table API or SQL query to an external system.

An input table can be registered from various sources:

* an existing `Table` object, usually the result of a Table API or SQL query.
* a `TableSource`, which accesses external data, such as a file, database, or messaging system.
* a `DataStream` or `DataSet` from a DataStream (only for stream job) or DataSet (only for batch job translated from old planner) program. Registering a `DataStream` or `DataSet` is discussed in the [Integration with DataStream and DataSet API](#integration-with-datastream-and-dataset-api) section.

An output table can be registered using a `TableSink`.

### Register a Table

A `Table` is registered in a `TableEnvironment` as follows:

{% highlight java %}
// get a TableEnvironment
TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

// table is the result of a simple projection query
Table projTable = tableEnv.scan("X").select(...);

// register the Table projTable as table "projectedTable"
tableEnv.registerTable("projectedTable", projTable);
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv = ... // see "Create a TableEnvironment" section

// table is the result of a simple projection query
val projTable: Table = tableEnv.scan("X").select(...)

// register the Table projTable as table "projectedTable"
tableEnv.registerTable("projectedTable", projTable)
{% endhighlight %}

{% highlight python %}
# get a TableEnvironment
table_env = ... # see "Create a TableEnvironment" section

# table is the result of a simple projection query
proj_table = table_env.scan("X").select(...)

# register the Table projTable as table "projectedTable"
table_env.register_table("projectedTable", proj_table)
{% endhighlight %}

**Note:** A registered `Table` is treated similarly to a `VIEW` as known from relational database systems, i.e., the query that defines the `Table` is not optimized but will be inlined when another query references the registered `Table`. If multiple queries reference the same registered `Table`, it will be inlined for each referencing query and executed multiple times, i.e., the result of the registered `Table` will *not* be shared.

{% top %}

### Register a TableSource

A `TableSource` provides access to external data which is stored in a storage system such as a database (MySQL, HBase, ...), a file with a specific encoding (CSV, Apache \[Parquet, Avro, ORC\], ...), or a messaging system (Apache Kafka, RabbitMQ, ...).

Flink aims to provide TableSources for common data formats and storage systems. Please have a look at the [Table Sources and Sinks]({{ site.baseurl }}/dev/table/sourceSinks.html) page for a list of supported TableSources and instructions for how to build a custom `TableSource`.

A `TableSource` is registered in a `TableEnvironment` as follows:

{% highlight java %}
// get a TableEnvironment
TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

// create a TableSource
TableSource csvSource = new CsvTableSource("/path/to/file", ...);

// register the TableSource as table "CsvTable"
tableEnv.registerTableSource("CsvTable", csvSource);
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv = ... // see "Create a TableEnvironment" section

// create a TableSource
val csvSource: TableSource = new CsvTableSource("/path/to/file", ...)

// register the TableSource as table "CsvTable"
tableEnv.registerTableSource("CsvTable", csvSource)
{% endhighlight %}

{% highlight python %}
# get a TableEnvironment
table_env = ... # see "Create a TableEnvironment" section

# create a TableSource
csv_source = CsvTableSource("/path/to/file", ...)

# register the TableSource as table "csvTable"
table_env.register_table_source("csvTable", csv_source)
{% endhighlight %}

**Note:** A `TableEnvironment` used for Blink planner only accepts `StreamTableSource`, `LookupableTableSource` and `InputFormatTableSource`, and a `StreamTableSource` used for Blink planner on batch must be bounded.

{% top %}

### Register a TableSink

A registered `TableSink` can be used to [emit the result of a Table API or SQL query](common.html#emit-a-table) to an external storage system, such as a database, key-value store, message queue, or file system (in different encodings, e.g., CSV, Apache \[Parquet, Avro, ORC\], ...).

Flink aims to provide TableSinks for common data formats and storage systems. Please see the documentation about [Table Sources and Sinks]({{ site.baseurl }}/dev/table/sourceSinks.html) page for details about available sinks and instructions for how to implement a custom `TableSink`.

A `TableSink` is registered in a `TableEnvironment` as follows:

{% highlight java %}
// get a TableEnvironment
TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

// create a TableSink
TableSink csvSink = new CsvTableSink("/path/to/file", ...);

// define the field names and types
String[] fieldNames = {"a", "b", "c"};
TypeInformation[] fieldTypes = {Types.INT, Types.STRING, Types.LONG};

// register the TableSink as table "CsvSinkTable"
tableEnv.registerTableSink("CsvSinkTable", fieldNames, fieldTypes, csvSink);
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv = ... // see "Create a TableEnvironment" section

// create a TableSink
val csvSink: TableSink = new CsvTableSink("/path/to/file", ...)

// define the field names and types
val fieldNames: Array[String] = Array("a", "b", "c")
val fieldTypes: Array[TypeInformation[_]] = Array(Types.INT, Types.STRING, Types.LONG)

// register the TableSink as table "CsvSinkTable"
tableEnv.registerTableSink("CsvSinkTable", fieldNames, fieldTypes, csvSink)
{% endhighlight %}

{% highlight python %}
# get a TableEnvironment
table_env = ... # see "Create a TableEnvironment" section

# define the field names and types
field_names = ["a", "b", "c"]
field_types = [DataTypes.INT(), DataTypes.STRING(), DataTypes.BIGINT()]

# create a TableSink
csv_sink = CsvTableSink(field_names, field_types, "/path/to/file", ...)

# register the TableSink as table "CsvSinkTable"
table_env.register_table_sink("CsvSinkTable", csv_sink)
{% endhighlight %}

{% top %}

Query a Table
-------------

### Table API

The Table API is a language-integrated query API for Scala and Java. In contrast to SQL, queries are not specified as Strings but are composed step-by-step in the host language.

The API is based on the `Table` class which represents a table (streaming or batch) and offers methods to apply relational operations. These methods return a new `Table` object, which represents the result of applying the relational operation on the input `Table`. Some relational operations are composed of multiple method calls such as `table.groupBy(...).select()`, where `groupBy(...)` specifies a grouping of `table`, and `select(...)` the projection on the grouping of `table`.

The [Table API]({{ site.baseurl }}/dev/table/tableApi.html) document describes all Table API operations that are supported on streaming and batch tables.

The following example shows a simple Table API aggregation query:

{% highlight java %}
// get a TableEnvironment
TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

// register Orders table

// scan registered Orders table
Table orders = tableEnv.scan("Orders");
// compute revenue for all customers from France
Table revenue = orders
  .filter("cCountry === 'FRANCE'")
  .groupBy("cID, cName")
  .select("cID, cName, revenue.sum AS revSum");

// emit or convert Table
// execute query
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv = ... // see "Create a TableEnvironment" section

// register Orders table

// scan registered Orders table
val orders = tableEnv.scan("Orders")
// compute revenue for all customers from France
val revenue = orders
  .filter('cCountry === "FRANCE")
  .groupBy('cID, 'cName)
  .select('cID, 'cName, 'revenue.sum AS 'revSum)

// emit or convert Table
// execute query
{% endhighlight %}

**Note:** The Scala Table API uses Scala Symbols, which start with a single tick (`'`) to reference the attributes of a `Table`. The Table API uses Scala implicits. Make sure to import `org.apache.flink.api.scala._` and `org.apache.flink.table.api.scala._` in order to use Scala implicit conversions.

{% highlight python %}
# get a TableEnvironment
table_env = # see "Create a TableEnvironment" section

# register Orders table

# scan registered Orders table
orders = table_env.scan("Orders")
# compute revenue for all customers from France
revenue = orders \
    .filter("cCountry === 'FRANCE'") \
    .group_by("cID, cName") \
    .select("cID, cName, revenue.sum AS revSum")

# emit or convert Table
# execute query
{% endhighlight %}

{% top %}

### SQL

Flink's SQL integration is based on [Apache Calcite](https://calcite.apache.org), which implements the SQL standard. SQL queries are specified as regular Strings.

The [SQL]({{ site.baseurl }}/dev/table/sql.html) document describes Flink's SQL support for streaming and batch tables.

The following example shows how to specify a query and return the result as a `Table`.

{% highlight java %}
// get a TableEnvironment
TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

// register Orders table

// compute revenue for all customers from France
Table revenue = tableEnv.sqlQuery(
    "SELECT cID, cName, SUM(revenue) AS revSum " +
    "FROM Orders " +
    "WHERE cCountry = 'FRANCE' " +
    "GROUP BY cID, cName"
  );

// emit or convert Table
// execute query
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv = ... // see "Create a TableEnvironment" section

// register Orders table

// compute revenue for all customers from France
val revenue = tableEnv.sqlQuery("""
  |SELECT cID, cName, SUM(revenue) AS revSum
  |FROM Orders
  |WHERE cCountry = 'FRANCE'
  |GROUP BY cID, cName
  """.stripMargin)

// emit or convert Table
// execute query
{% endhighlight %}

{% highlight python %}
# get a TableEnvironment
table_env = ... # see "Create a TableEnvironment" section

# register Orders table

# compute revenue for all customers from France
revenue = table_env.sql_query(
    "SELECT cID, cName, SUM(revenue) AS revSum "
    "FROM Orders "
    "WHERE cCountry = 'FRANCE' "
    "GROUP BY cID, cName"
)

# emit or convert Table
# execute query
{% endhighlight %}

The following example shows how to specify an update query that inserts its result into a registered table.

{% highlight java %}
// get a TableEnvironment
TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

// register "Orders" table
// register "RevenueFrance" output table

// compute revenue for all customers from France and emit to "RevenueFrance"
tableEnv.sqlUpdate(
    "INSERT INTO RevenueFrance " +
    "SELECT cID, cName, SUM(revenue) AS revSum " +
    "FROM Orders " +
    "WHERE cCountry = 'FRANCE' " +
    "GROUP BY cID, cName"
  );

// execute query
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv = ... // see "Create a TableEnvironment" section

// register "Orders" table
// register "RevenueFrance" output table

// compute revenue for all customers from France and emit to "RevenueFrance"
tableEnv.sqlUpdate("""
  |INSERT INTO RevenueFrance
  |SELECT cID, cName, SUM(revenue) AS revSum
  |FROM Orders
  |WHERE cCountry = 'FRANCE'
  |GROUP BY cID, cName
  """.stripMargin)

// execute query
{% endhighlight %}

{% highlight python %}
# get a TableEnvironment
table_env = ... # see "Create a TableEnvironment" section

# register "Orders" table
# register "RevenueFrance" output table

# compute revenue for all customers from France and emit to "RevenueFrance"
table_env.sql_update(
    "INSERT INTO RevenueFrance "
    "SELECT cID, cName, SUM(revenue) AS revSum "
    "FROM Orders "
    "WHERE cCountry = 'FRANCE' "
    "GROUP BY cID, cName"
)

# execute query
{% endhighlight %}

{% top %}

### Mixing Table API and SQL

Table API and SQL queries can be easily mixed because both return `Table` objects:

* A Table API query can be defined on the `Table` object returned by a SQL query.
* A SQL query can be defined on the result of a Table API query by [registering the resulting Table](#register-a-table) in the `TableEnvironment` and referencing it in the `FROM` clause of the SQL query.

{% top %}

Emit a Table
------------

A `Table` is emitted by writing it to a `TableSink`. A `TableSink` is a generic interface to support a wide variety of file formats (e.g. CSV, Apache Parquet, Apache Avro), storage systems (e.g., JDBC, Apache HBase, Apache Cassandra, Elasticsearch), or messaging systems (e.g., Apache Kafka, RabbitMQ).

A batch `Table` can only be written to a `BatchTableSink`, while a streaming `Table` requires either an `AppendStreamTableSink`, a `RetractStreamTableSink`, or an `UpsertStreamTableSink`.

Please see the documentation about [Table Sources & Sinks]({{ site.baseurl }}/dev/table/sourceSinks.html) for details about available sinks and instructions for how to implement a custom `TableSink`.

The `Table.insertInto(String tableName)` method emits the `Table` to a registered `TableSink`. The method looks up the `TableSink` from the catalog by the name and validates that the schema of the `Table` is identical to the schema of the `TableSink`.

The following examples shows how to emit a `Table`:

{% highlight java %}
// get a TableEnvironment
TableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

// create a TableSink
TableSink sink = new CsvTableSink("/path/to/file", fieldDelim = "|");

// register the TableSink with a specific schema
String[] fieldNames = {"a", "b", "c"};
TypeInformation[] fieldTypes = {Types.INT, Types.STRING, Types.LONG};
tableEnv.registerTableSink("CsvSinkTable", fieldNames, fieldTypes, sink);

// compute a result Table using Table API operators and/or SQL queries
Table result = ...
// emit the result Table to the registered TableSink
result.insertInto("CsvSinkTable");

// execute the program
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv = ... // see "Create a TableEnvironment" section

// create a TableSink
val sink: TableSink = new CsvTableSink("/path/to/file", fieldDelim = "|")

// register the TableSink with a specific schema
val fieldNames: Array[String] = Array("a", "b", "c")
val fieldTypes: Array[TypeInformation] = Array(Types.INT, Types.STRING, Types.LONG)
tableEnv.registerTableSink("CsvSinkTable", fieldNames, fieldTypes, sink)

// compute a result Table using Table API operators and/or SQL queries
val result: Table = ...

// emit the result Table to the registered TableSink
result.insertInto("CsvSinkTable")

// execute the program
{% endhighlight %}

{% highlight python %}
# get a TableEnvironment
table_env = ... # see "Create a TableEnvironment" section

field_names = ["a", "b", "c"]
field_types = [DataTypes.INT(), DataTypes.STRING(), DataTypes.BIGINT()]

# create a TableSink
sink = CsvTableSink(field_names, field_types, "/path/to/file", "|")

table_env.register_table_sink("CsvSinkTable", sink)

# compute a result Table using Table API operators and/or SQL queries
result = ...

# emit the result Table to the registered TableSink
result.insert_into("CsvSinkTable")

# execute the program
{% endhighlight %}

{% top %}

Translate and Execute a Query
-----------------------------

The behavior of translating and executing a query is different for the two planners.

Table API and SQL queries are translated into [DataStream]({{ site.baseurl }}/dev/datastream_api.html) or [DataSet]({{ site.baseurl }}/dev/batch) programs depending on whether their input is a streaming or batch input. A query is internally represented as a logical query plan and is translated in two phases:

1. Optimization of the logical plan
2. Translation into a DataStream or DataSet program

A Table API or SQL query is translated when:

* a `Table` is emitted to a `TableSink`, i.e., when `Table.insertInto()` is called.
* a SQL update query is specified, i.e., when `TableEnvironment.sqlUpdate()` is called.
* a `Table` is converted into a `DataStream` or `DataSet` (see [Integration with DataStream and DataSet API](#integration-with-datastream-and-dataset-api)).

Once translated, a Table API or SQL query is handled like a regular DataStream or DataSet program and is executed when `StreamExecutionEnvironment.execute()` or `ExecutionEnvironment.execute()` is called.

Table API and SQL queries are translated into [DataStream]({{ site.baseurl }}/dev/datastream_api.html) programs whether their input is streaming or batch. A query is internally represented as a logical query plan and is translated in two phases:

1. Optimization of the logical plan,
2. Translation into a DataStream program.

The behavior of translating  a query is different for `TableEnvironment` and `StreamTableEnvironment`.

For `TableEnvironment`, A Table API or SQL query is translated when `TableEnvironment.execute()` is called, because `TableEnvironment` will optimize multiple-sinks into one DAG.

while for `StreamTableEnvironment`, A Table API or SQL query is translated when:

* a `Table` is emitted to a `TableSink`, i.e., when `Table.insertInto()` is called.
* a SQL update query is specified, i.e., when `TableEnvironment.sqlUpdate()` is called.
* a `Table` is converted into a `DataStream`.

Once translated, a Table API or SQL query is handled like a regular DataStream program and is executed when `TableEnvironment.execute()` or `StreamExecutionEnvironment.execute()` is called.

{% top %}

Integration with DataStream and DataSet API
-------------------------------------------

Both planners on stream can integrate with the `DataStream` API. Only old planner can integrate with the `DataSet API`, Blink planner on batch could not be combined with both.
**Note:** The `DataSet` API discussed below is only relevant for the old planner on batch.

Table API and SQL queries can be easily integrated with and embedded into [DataStream]({{ site.baseurl }}/dev/datastream_api.html) and [DataSet]({{ site.baseurl }}/dev/batch) programs. For instance, it is possible to query an external table (for example from a RDBMS), do some pre-processing, such as filtering, projecting, aggregating, or joining with meta data, and then further process the data with either the DataStream or DataSet API (and any of the libraries built on top of these APIs, such as CEP or Gelly). Inversely, a Table API or SQL query can also be applied on the result of a DataStream or DataSet program.

This interaction can be achieved by converting a `DataStream` or `DataSet` into a `Table` and vice versa. In this section, we describe how these conversions are done.

### Implicit Conversion for Scala

The Scala Table API features implicit conversions for the `DataSet`, `DataStream`, and `Table` classes. These conversions are enabled by importing the package `org.apache.flink.table.api.scala._` in addition to `org.apache.flink.api.scala._` for the Scala DataStream API.

### Register a DataStream or DataSet as Table

A `DataStream` or `DataSet` can be registered in a `TableEnvironment` as a Table. The schema of the resulting table depends on the data type of the registered `DataStream` or `DataSet`. Please check the section about [mapping of data types to table schema](#mapping-of-data-types-to-table-schema) for details.

{% highlight java %}
// get StreamTableEnvironment
// registration of a DataSet in a BatchTableEnvironment is equivalent
StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

DataStream<Tuple2<Long, String>> stream = ...

// register the DataStream as Table "myTable" with fields "f0", "f1"
tableEnv.registerDataStream("myTable", stream);

// register the DataStream as table "myTable2" with fields "myLong", "myString"
tableEnv.registerDataStream("myTable2", stream, "myLong, myString");
{% endhighlight %}

{% highlight scala %}
// get TableEnvironment
// registration of a DataSet is equivalent
val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section

val stream: DataStream[(Long, String)] = ...

// register the DataStream as Table "myTable" with fields "f0", "f1"
tableEnv.registerDataStream("myTable", stream)

// register the DataStream as table "myTable2" with fields "myLong", "myString"
tableEnv.registerDataStream("myTable2", stream, 'myLong, 'myString)
{% endhighlight %}

**Note:** The name of a `DataStream` `Table` must not match the `^_DataStreamTable_[0-9]+` pattern and the name of a `DataSet` `Table` must not match the `^_DataSetTable_[0-9]+` pattern. These patterns are reserved for internal use only.

{% top %}

### Convert a DataStream or DataSet into a Table

Instead of registering a `DataStream` or `DataSet` in a `TableEnvironment`, it can also be directly converted into a `Table`. This is convenient if you want to use the Table in a Table API query.

{% highlight java %}
// get StreamTableEnvironment
// registration of a DataSet in a BatchTableEnvironment is equivalent
StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

DataStream<Tuple2<Long, String>> stream = ...

// Convert the DataStream into a Table with default fields "f0", "f1"
Table table1 = tableEnv.fromDataStream(stream);

// Convert the DataStream into a Table with fields "myLong", "myString"
Table table2 = tableEnv.fromDataStream(stream, "myLong, myString");
{% endhighlight %}

{% highlight scala %}
// get TableEnvironment
// registration of a DataSet is equivalent
val tableEnv = ... // see "Create a TableEnvironment" section

val stream: DataStream[(Long, String)] = ...

// convert the DataStream into a Table with default fields '_1, '_2
val table1: Table = tableEnv.fromDataStream(stream)

// convert the DataStream into a Table with fields 'myLong, 'myString
val table2: Table = tableEnv.fromDataStream(stream, 'myLong, 'myString)
{% endhighlight %}

{% top %}

### Convert a Table into a DataStream or DataSet

A `Table` can be converted into a `DataStream` or `DataSet`. In this way, custom DataStream or DataSet programs can be run on the result of a Table API or SQL query.

When converting a `Table` into a `DataStream` or `DataSet`, you need to specify the data type of the resulting `DataStream` or `DataSet`, i.e., the data type into which the rows of the `Table` are to be converted. Often the most convenient conversion type is `Row`. The following list gives an overview of the features of the different options:

- **Row**: fields are mapped by position, arbitrary number of fields, support for `null` values, no type-safe access.
- **POJO**: fields are mapped by name (POJO fields must be named as `Table` fields), arbitrary number of fields, support for `null` values, type-safe access.
- **Case Class**: fields are mapped by position, no support for `null` values, type-safe access.
- **Tuple**: fields are mapped by position, limitation to 22 (Scala) or 25 (Java) fields, no support for `null` values, type-safe access.
- **Atomic Type**: `Table` must have a single field, no support for `null` values, type-safe access.

#### Convert a Table into a DataStream

A `Table` that is the result of a streaming query will be updated dynamically, i.e., it is changing as new records arrive on the query's input streams. Hence, the `DataStream` into which such a dynamic query is converted needs to encode the updates of the table.

There are two modes to convert a `Table` into a `DataStream`:

1. **Append Mode**: This mode can only be used if the dynamic `Table` is only modified by `INSERT` changes, i.e, it is append-only and previously emitted results are never updated.
2. **Retract Mode**: This mode can always be used. It encodes `INSERT` and `DELETE` changes with a `boolean` flag.

{% highlight java %}
// get StreamTableEnvironment.
StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

// Table with two fields (String name, Integer age)
Table table = ...

// convert the Table into an append DataStream of Row by specifying the class
DataStream<Row> dsRow = tableEnv.toAppendStream(table, Row.class);

// convert the Table into an append DataStream of Tuple2<String, Integer>
//   via a TypeInformation
TupleTypeInfo<Tuple2<String, Integer>> tupleType = new TupleTypeInfo<>(
  Types.STRING(),
  Types.INT());
DataStream<Tuple2<String, Integer>> dsTuple =
  tableEnv.toAppendStream(table, tupleType);

// convert the Table into a retract DataStream of Row.
//   A retract stream of type X is a DataStream<Tuple2<Boolean, X>>.
//   The boolean field indicates the type of the change.
//   True is INSERT, false is DELETE.
DataStream<Tuple2<Boolean, Row>> retractStream =
  tableEnv.toRetractStream(table, Row.class);

{% endhighlight %}

{% highlight scala %}
// get TableEnvironment.
// registration of a DataSet is equivalent
val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section

// Table with two fields (String name, Integer age)
val table: Table = ...

// convert the Table into an append DataStream of Row
val dsRow: DataStream[Row] = tableEnv.toAppendStream[Row](table)

// convert the Table into an append DataStream of Tuple2[String, Int]
val dsTuple: DataStream[(String, Int)] dsTuple =
  tableEnv.toAppendStream[(String, Int)](table)

// convert the Table into a retract DataStream of Row.
//   A retract stream of type X is a DataStream[(Boolean, X)].
//   The boolean field indicates the type of the change.
//   True is INSERT, false is DELETE.
val retractStream: DataStream[(Boolean, Row)] = tableEnv.toRetractStream[Row](table)
{% endhighlight %}

**Note:** A detailed discussion about dynamic tables and their properties is given in the [Dynamic Tables](streaming/dynamic_tables.html) document.

#### Convert a Table into a DataSet

A `Table` is converted into a `DataSet` as follows:

{% highlight java %}
// get BatchTableEnvironment
BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env);

// Table with two fields (String name, Integer age)
Table table = ...

// convert the Table into a DataSet of Row by specifying a class
DataSet<Row> dsRow = tableEnv.toDataSet(table, Row.class);

// convert the Table into a DataSet of Tuple2<String, Integer> via a TypeInformation
TupleTypeInfo<Tuple2<String, Integer>> tupleType = new TupleTypeInfo<>(
  Types.STRING(),
  Types.INT());
DataSet<Tuple2<String, Integer>> dsTuple =
  tableEnv.toDataSet(table, tupleType);
{% endhighlight %}

{% highlight scala %}
// get TableEnvironment
// registration of a DataSet is equivalent
val tableEnv = BatchTableEnvironment.create(env)

// Table with two fields (String name, Integer age)
val table: Table = ...

// convert the Table into a DataSet of Row
val dsRow: DataSet[Row] = tableEnv.toDataSet[Row](table)

// convert the Table into a DataSet of Tuple2[String, Int]
val dsTuple: DataSet[(String, Int)] = tableEnv.toDataSet[(String, Int)](table)
{% endhighlight %}

{% top %}

### Mapping of Data Types to Table Schema

Flink's DataStream and DataSet APIs support very diverse types. Composite types such as Tuples (built-in Scala and Flink Java tuples), POJOs, Scala case classes, and Flink's Row type allow for nested data structures with multiple fields that can be accessed in table expressions. Other types are treated as atomic types. In the following, we describe how the Table API converts these types into an internal row representation and show examples of converting a `DataStream` into a `Table`.

The mapping of a data type to a table schema can happen in two ways: **based on the field positions** or **based on the field names**.

**Position-based Mapping**

Position-based mapping can be used to give fields a more meaningful name while keeping the field order. This mapping is available for composite data types *with a defined field order* as well as atomic types. Composite data types such as tuples, rows, and case classes have such a field order. However, fields of a POJO must be mapped based on the field names (see next section). Fields can be projected out but can't be renamed using an alias `as`.

When defining a position-based mapping, the specified names must not exist in the input data type, otherwise the API will assume that the mapping should happen based on the field names. If no field names are specified, the default field names and field order of the composite type are used or `f0` for atomic types.

{% highlight java %}
// get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section;

DataStream<Tuple2<Long, Integer>> stream = ...

// convert DataStream into Table with default field names "f0" and "f1"
Table table = tableEnv.fromDataStream(stream);

// convert DataStream into Table with field "myLong" only
Table table = tableEnv.fromDataStream(stream, "myLong");

// convert DataStream into Table with field names "myLong" and "myInt"
Table table = tableEnv.fromDataStream(stream, "myLong, myInt");
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section

val stream: DataStream[(Long, Int)] = ...

// convert DataStream into Table with default field names "_1" and "_2"
val table: Table = tableEnv.fromDataStream(stream)

// convert DataStream into Table with field "myLong" only
val table: Table = tableEnv.fromDataStream(stream, 'myLong)

// convert DataStream into Table with field names "myLong" and "myInt"
val table: Table = tableEnv.fromDataStream(stream, 'myLong, 'myInt)
{% endhighlight %}

**Name-based Mapping**

Name-based mapping can be used for any data type including POJOs. It is the most flexible way of defining a table schema mapping. All fields in the mapping are referenced by name and can be possibly renamed using an alias `as`. Fields can be reordered and projected out.

If no field names are specified, the default field names and field order of the composite type are used or `f0` for atomic types.

{% highlight java %}
// get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

DataStream<Tuple2<Long, Integer>> stream = ...

// convert DataStream into Table with default field names "f0" and "f1"
Table table = tableEnv.fromDataStream(stream);

// convert DataStream into Table with field "f1" only
Table table = tableEnv.fromDataStream(stream, "f1");

// convert DataStream into Table with swapped fields
Table table = tableEnv.fromDataStream(stream, "f1, f0");

// convert DataStream into Table with swapped fields and field names "myInt" and "myLong"
Table table = tableEnv.fromDataStream(stream, "f1 as myInt, f0 as myLong");
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section

val stream: DataStream[(Long, Int)] = ...

// convert DataStream into Table with default field names "_1" and "_2"
val table: Table = tableEnv.fromDataStream(stream)

// convert DataStream into Table with field "_2" only
val table: Table = tableEnv.fromDataStream(stream, '_2)

// convert DataStream into Table with swapped fields
val table: Table = tableEnv.fromDataStream(stream, '_2, '_1)

// convert DataStream into Table with swapped fields and field names "myInt" and "myLong"
val table: Table = tableEnv.fromDataStream(stream, '_2 as 'myInt, '_1 as 'myLong)
{% endhighlight %}

#### Atomic Types

Flink treats primitives (`Integer`, `Double`, `String`) or generic types (types that cannot be analyzed and decomposed) as atomic types. A `DataStream` or `DataSet` of an atomic type is converted into a `Table` with a single attribute. The type of the attribute is inferred from the atomic type and the name of the attribute can be specified.

{% highlight java %}
// get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

DataStream<Long> stream = ...

// convert DataStream into Table with default field name "f0"
Table table = tableEnv.fromDataStream(stream);

// convert DataStream into Table with field name "myLong"
Table table = tableEnv.fromDataStream(stream, "myLong");
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section

val stream: DataStream[Long] = ...

// convert DataStream into Table with default field name "f0"
val table: Table = tableEnv.fromDataStream(stream)

// convert DataStream into Table with field name "myLong"
val table: Table = tableEnv.fromDataStream(stream, 'myLong)
{% endhighlight %}

#### Tuples (Scala and Java) and Case Classes (Scala only)

Flink supports Scala's built-in tuples and provides its own tuple classes for Java. DataStreams and DataSets of both kinds of 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 (`f0`, `f1`, ... for Flink Tuples and `_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`).

{% highlight java %}
// get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

DataStream<Tuple2<Long, String>> stream = ...

// convert DataStream into Table with default field names "f0", "f1"
Table table = tableEnv.fromDataStream(stream);

// convert DataStream into Table with renamed field names "myLong", "myString" (position-based)
Table table = tableEnv.fromDataStream(stream, "myLong, myString");

// convert DataStream into Table with reordered fields "f1", "f0" (name-based)
Table table = tableEnv.fromDataStream(stream, "f1, f0");

// convert DataStream into Table with projected field "f1" (name-based)
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'");
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section

val stream: DataStream[(Long, String)] = ...

// convert DataStream into Table with renamed default field names '_1, '_2
val table: Table = tableEnv.fromDataStream(stream)

// 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 default field names 'name, 'age
val table = tableEnv.fromDataStream(streamCC)

// 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)

{% endhighlight %}

#### POJO (Java and Scala)

Flink supports POJOs as composite types. The rules for what determines a POJO are documented [here]({{ site.baseurl }}/dev/api_concepts.html#pojos).

When converting a POJO `DataStream` or `DataSet` into a `Table` without specifying field names, the names of the original POJO fields are used. The name mapping requires the original names and cannot be done by positions. Fields can be renamed using an alias (with the `as` keyword), reordered, and projected.

{% highlight java %}
// get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

// Person is a POJO with fields "name" and "age"
DataStream<Person> stream = ...

// convert DataStream into Table with default field names "age", "name" (fields are ordered by name!)
Table table = tableEnv.fromDataStream(stream);

// convert DataStream into Table with renamed fields "myAge", "myName" (name-based)
Table table = tableEnv.fromDataStream(stream, "age as myAge, name as myName");

// convert DataStream into Table with projected field "name" (name-based)
Table table = tableEnv.fromDataStream(stream, "name");

// convert DataStream into Table with projected and renamed field "myName" (name-based)
Table table = tableEnv.fromDataStream(stream, "name as myName");
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section

// Person is a POJO with field names "name" and "age"
val stream: DataStream[Person] = ...

// convert DataStream into Table with default field names "age", "name" (fields are ordered by name!)
val table: Table = tableEnv.fromDataStream(stream)

// convert DataStream into Table with renamed fields "myAge", "myName" (name-based)
val table: Table = tableEnv.fromDataStream(stream, 'age as 'myAge, 'name as 'myName)

// convert DataStream into Table with projected field "name" (name-based)
val table: Table = tableEnv.fromDataStream(stream, 'name)

// convert DataStream into Table with projected and renamed field "myName" (name-based)
val table: Table = tableEnv.fromDataStream(stream, 'name as 'myName)
{% endhighlight %}

#### Row

The `Row` data type supports an arbitrary number of fields and fields with `null` values. Field names can be specified via a `RowTypeInfo` or when converting a `Row` `DataStream` or `DataSet` into a `Table`. The row type supports mapping of fields by position and by name. Fields can be renamed by providing names for all fields (mapping based on position) or selected individually for projection/ordering/renaming (mapping based on name).

{% highlight java %}
// get a StreamTableEnvironment, works for BatchTableEnvironment equivalently
StreamTableEnvironment tableEnv = ...; // see "Create a TableEnvironment" section

// DataStream of Row with two fields "name" and "age" specified in `RowTypeInfo`
DataStream<Row> stream = ...

// convert DataStream into Table with default field names "name", "age"
Table table = tableEnv.fromDataStream(stream);

// convert DataStream into Table with renamed field names "myName", "myAge" (position-based)
Table table = tableEnv.fromDataStream(stream, "myName, myAge");

// convert DataStream into Table with renamed fields "myName", "myAge" (name-based)
Table table = tableEnv.fromDataStream(stream, "name as myName, age as myAge");

// convert DataStream into Table with projected field "name" (name-based)
Table table = tableEnv.fromDataStream(stream, "name");

// convert DataStream into Table with projected and renamed field "myName" (name-based)
Table table = tableEnv.fromDataStream(stream, "name as myName");
{% endhighlight %}

{% highlight scala %}
// get a TableEnvironment
val tableEnv: StreamTableEnvironment = ... // see "Create a TableEnvironment" section

// DataStream of Row with two fields "name" and "age" specified in `RowTypeInfo`
val stream: DataStream[Row] = ...

// convert DataStream into Table with default field names "name", "age"
val table: Table = tableEnv.fromDataStream(stream)

// convert DataStream into Table with renamed field names "myName", "myAge" (position-based)
val table: Table = tableEnv.fromDataStream(stream, 'myName, 'myAge)

// convert DataStream into Table with renamed fields "myName", "myAge" (name-based)
val table: Table = tableEnv.fromDataStream(stream, 'name as 'myName, 'age as 'myAge)

// convert DataStream into Table with projected field "name" (name-based)
val table: Table = tableEnv.fromDataStream(stream, 'name)

// convert DataStream into Table with projected and renamed field "myName" (name-based)
val table: Table = tableEnv.fromDataStream(stream, 'name as 'myName)
{% endhighlight %}

{% top %}

Query Optimization
------------------

Apache Flink leverages Apache Calcite to optimize and translate queries. The optimization currently performed include projection and filter push-down, subquery decorrelation, and other kinds of query rewriting. Old planner does not yet optimize the order of joins, but executes them in the same order as defined in the query (order of Tables in the `FROM` clause and/or order of join predicates in the `WHERE` clause).

It is possible to tweak the set of optimization rules which are applied in different phases by providing a `CalciteConfig` object. This can be created via a builder by calling `CalciteConfig.createBuilder())` and is provided to the TableEnvironment by calling `tableEnv.getConfig.setPlannerConfig(calciteConfig)`.

Apache Flink leverages and extends Apache Calcite to perform sophisticated query optimization.
This includes a series of rule and cost-based optimizations such as:

* Subquery decorrelation based on Apache Calcite
* Project pruning
* Partition pruning
* Filter push-down
* Sub-plan deduplication to avoid duplicate computation
* Special subquery rewriting, including two parts:
    * Converts IN and EXISTS into left semi-joins
    * Converts NOT IN and NOT EXISTS into left anti-join
* Optional join reordering
    * Enabled via `table.optimizer.join-reorder-enabled`

**Note:** IN/EXISTS/NOT IN/NOT EXISTS are currently only supported in conjunctive conditions in subquery rewriting.

The optimizer makes intelligent decisions, based not only on the plan but also rich statistics available from the data sources and fine-grain costs for each operator such as io, cpu, network, and memory.

Advanced users may provide custom optimizations via a `CalciteConfig` object that can be provided to the table environment by calling `TableEnvironment#getConfig#setPlannerConfig`.

### Explaining a Table

The Table API provides a mechanism to explain the logical and optimized query plans to compute a `Table`.
This is done through the `TableEnvironment.explain(table)` method or `TableEnvironment.explain()` method. `explain(table)` returns the plan of a given `Table`. `explain()` returns the result of a multiple-sinks plan and is mainly used for the Blink planner. It returns a String describing three plans:

1. the Abstract Syntax Tree of the relational query, i.e., the unoptimized logical query plan,
2. the optimized logical query plan, and
3. the physical execution plan.

The following code shows an example and the corresponding output for given `Table` using `explain(table)`:

{% highlight java %}
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

DataStream<Tuple2<Integer, String>> stream1 = env.fromElements(new Tuple2<>(1, "hello"));
DataStream<Tuple2<Integer, String>> stream2 = env.fromElements(new Tuple2<>(1, "hello"));

Table table1 = tEnv.fromDataStream(stream1, "count, word");
Table table2 = tEnv.fromDataStream(stream2, "count, word");
Table table = table1
  .where("LIKE(word, 'F%')")
  .unionAll(table2);

String explanation = tEnv.explain(table);
System.out.println(explanation);
{% endhighlight %}

{% highlight scala %}
val env = StreamExecutionEnvironment.getExecutionEnvironment
val tEnv = StreamTableEnvironment.create(env)

val table1 = env.fromElements((1, "hello")).toTable(tEnv, 'count, 'word)
val table2 = env.fromElements((1, "hello")).toTable(tEnv, 'count, 'word)
val table = table1
  .where('word.like("F%"))
  .unionAll(table2)

val explanation: String = tEnv.explain(table)
println(explanation)
{% endhighlight %}

{% highlight python %}
env = StreamExecutionEnvironment.get_execution_environment()
t_env = StreamTableEnvironment.create(env)

table1 = t_env.from_elements([(1, "hello")], ["count", "word"])
table2 = t_env.from_elements([(1, "hello")], ["count", "word"])
table = table1 \
    .where("LIKE(word, 'F%')") \
    .union_all(table2)

explanation = t_env.explain(table)
print(explanation)
{% endhighlight %}

{% highlight text %}
== Abstract Syntax Tree ==
LogicalUnion(all=[true])
  LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
    FlinkLogicalDataStreamScan(id=[1], fields=[count, word])
  FlinkLogicalDataStreamScan(id=[2], fields=[count, word])

== Optimized Logical Plan ==
DataStreamUnion(all=[true], union all=[count, word])
  DataStreamCalc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
    DataStreamScan(id=[1], fields=[count, word])
  DataStreamScan(id=[2], fields=[count, word])

== Physical Execution Plan ==
Stage 1 : Data Source
	content : collect elements with CollectionInputFormat

Stage 2 : Data Source
	content : collect elements with CollectionInputFormat

	Stage 3 : Operator
		content : from: (count, word)
		ship_strategy : REBALANCE

		Stage 4 : Operator
			content : where: (LIKE(word, _UTF-16LE'F%')), select: (count, word)
			ship_strategy : FORWARD

			Stage 5 : Operator
				content : from: (count, word)
				ship_strategy : REBALANCE
{% endhighlight %}

{% highlight text %}
== Abstract Syntax Tree ==
LogicalUnion(all=[true])
  LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
    FlinkLogicalDataStreamScan(id=[1], fields=[count, word])
  FlinkLogicalDataStreamScan(id=[2], fields=[count, word])

== Optimized Logical Plan ==
DataStreamUnion(all=[true], union all=[count, word])
  DataStreamCalc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
    DataStreamScan(id=[1], fields=[count, word])
  DataStreamScan(id=[2], fields=[count, word])

== Physical Execution Plan ==
Stage 1 : Data Source
	content : collect elements with CollectionInputFormat

Stage 2 : Data Source
	content : collect elements with CollectionInputFormat

	Stage 3 : Operator
		content : from: (count, word)
		ship_strategy : REBALANCE

		Stage 4 : Operator
			content : where: (LIKE(word, _UTF-16LE'F%')), select: (count, word)
			ship_strategy : FORWARD

			Stage 5 : Operator
				content : from: (count, word)
				ship_strategy : REBALANCE
{% endhighlight %}

{% highlight text %}
== Abstract Syntax Tree ==
LogicalUnion(all=[true])
  LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
    FlinkLogicalDataStreamScan(id=[3], fields=[count, word])
  FlinkLogicalDataStreamScan(id=[6], fields=[count, word])

== Optimized Logical Plan ==
DataStreamUnion(all=[true], union all=[count, word])
  DataStreamCalc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')])
    DataStreamScan(id=[3], fields=[count, word])
  DataStreamScan(id=[6], fields=[count, word])

== Physical Execution Plan ==
Stage 1 : Data Source
	content : collect elements with CollectionInputFormat

	Stage 2 : Operator
		content : Flat Map
		ship_strategy : FORWARD

		Stage 3 : Operator
			content : Map
			ship_strategy : FORWARD

Stage 4 : Data Source
	content : collect elements with CollectionInputFormat

	Stage 5 : Operator
		content : Flat Map
		ship_strategy : FORWARD

		Stage 6 : Operator
			content : Map
			ship_strategy : FORWARD

			Stage 7 : Operator
				content : Map
				ship_strategy : FORWARD

				Stage 8 : Operator
					content : where: (LIKE(word, _UTF-16LE'F%')), select: (count, word)
					ship_strategy : FORWARD

					Stage 9 : Operator
						content : Map
						ship_strategy : FORWARD
{% endhighlight %}

The following code shows an example and the corresponding output for multiple-sinks plan using `explain()`:

{% highlight java %}

EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
TableEnvironment tEnv = TableEnvironment.create(settings);

String[] fieldNames = { "count", "word" };
TypeInformation[] fieldTypes = { Types.INT, Types.STRING };
tEnv.registerTableSource("MySource1", new CsvTableSource("/source/path1", fieldNames, fieldTypes));
tEnv.registerTableSource("MySource2", new CsvTableSource("/source/path2", fieldNames, fieldTypes));
tEnv.registerTableSink("MySink1", new CsvTableSink("/sink/path1").configure(fieldNames, fieldTypes));
tEnv.registerTableSink("MySink2", new CsvTableSink("/sink/path2").configure(fieldNames, fieldTypes));

Table table1 = tEnv.scan("MySource1").where("LIKE(word, 'F%')");
table1.insertInto("MySink1");

Table table2 = table1.unionAll(tEnv.scan("MySource2"));
table2.insertInto("MySink2");

String explanation = tEnv.explain(false);
System.out.println(explanation);

{% endhighlight %}

{% highlight scala %}
val settings = EnvironmentSettings.newInstance.useBlinkPlanner.inStreamingMode.build
val tEnv = TableEnvironment.create(settings)

val fieldNames = Array("count", "word")
val fieldTypes = Array[TypeInformation[_]](Types.INT, Types.STRING)
tEnv.registerTableSource("MySource1", new CsvTableSource("/source/path1", fieldNames, fieldTypes))
tEnv.registerTableSource("MySource2", new CsvTableSource("/source/path2",fieldNames, fieldTypes))
tEnv.registerTableSink("MySink1", new CsvTableSink("/sink/path1").configure(fieldNames, fieldTypes))
tEnv.registerTableSink("MySink2", new CsvTableSink("/sink/path2").configure(fieldNames, fieldTypes))

val table1 = tEnv.scan("MySource1").where("LIKE(word, 'F%')")
table1.insertInto("MySink1")

val table2 = table1.unionAll(tEnv.scan("MySource2"))
table2.insertInto("MySink2")

val explanation = tEnv.explain(false)
println(explanation)

{% endhighlight %}

{% highlight python %}
settings = EnvironmentSettings.new_instance().use_blink_planner().in_streaming_mode().build()
t_env = TableEnvironment.create(environment_settings=settings)

field_names = ["count", "word"]
field_types = [DataTypes.INT(), DataTypes.STRING()]
t_env.register_table_source("MySource1", CsvTableSource("/source/path1", field_names, field_types))
t_env.register_table_source("MySource2", CsvTableSource("/source/path2", field_names, field_types))
t_env.register_table_sink("MySink1", CsvTableSink("/sink/path1", field_names, field_types))
t_env.register_table_sink("MySink2", CsvTableSink("/sink/path2", field_names, field_types))

table1 = t_env.scan("MySource1").where("LIKE(word, 'F%')")
table1.insert_into("MySink1")

table2 = table1.union_all(t_env.scan("MySource2"))
table2.insert_into("MySink2")

explanation = t_env.explain()
print(explanation)
{% endhighlight %}

the result of multiple-sinks plan is

{% highlight text %}

== Abstract Syntax Tree ==
LogicalSink(name=[MySink1], fields=[count, word])
+- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
   +- LogicalTableScan(table=[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]])

LogicalSink(name=[MySink2], fields=[count, word])
+- LogicalUnion(all=[true])
   :- LogicalFilter(condition=[LIKE($1, _UTF-16LE'F%')])
   :  +- LogicalTableScan(table=[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]])
   +- LogicalTableScan(table=[[default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]]])

== Optimized Logical Plan ==
Calc(select=[count, word], where=[LIKE(word, _UTF-16LE'F%')], reuse_id=[1])
+- TableSourceScan(table=[[default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]]], fields=[count, word])

Sink(name=[MySink1], fields=[count, word])
+- Reused(reference_id=[1])

Sink(name=[MySink2], fields=[count, word])
+- Union(all=[true], union=[count, word])
   :- Reused(reference_id=[1])
   +- TableSourceScan(table=[[default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]]], fields=[count, word])

== Physical Execution Plan ==
Stage 1 : Data Source
	content : collect elements with CollectionInputFormat

	Stage 2 : Operator
		content : CsvTableSource(read fields: count, word)
		ship_strategy : REBALANCE

		Stage 3 : Operator
			content : SourceConversion(table:Buffer(default_catalog, default_database, MySource1, source: [CsvTableSource(read fields: count, word)]), fields:(count, word))
			ship_strategy : FORWARD

			Stage 4 : Operator
				content : Calc(where: (word LIKE _UTF-16LE'F%'), select: (count, word))
				ship_strategy : FORWARD

				Stage 5 : Operator
					content : SinkConversionToRow
					ship_strategy : FORWARD

					Stage 6 : Operator
						content : Map
						ship_strategy : FORWARD

Stage 8 : Data Source
	content : collect elements with CollectionInputFormat

	Stage 9 : Operator
		content : CsvTableSource(read fields: count, word)
		ship_strategy : REBALANCE

		Stage 10 : Operator
			content : SourceConversion(table:Buffer(default_catalog, default_database, MySource2, source: [CsvTableSource(read fields: count, word)]), fields:(count, word))
			ship_strategy : FORWARD

			Stage 12 : Operator
				content : SinkConversionToRow
				ship_strategy : FORWARD

				Stage 13 : Operator
					content : Map
					ship_strategy : FORWARD

					Stage 7 : Data Sink
						content : Sink: CsvTableSink(count, word)
						ship_strategy : FORWARD

						Stage 14 : Data Sink
							content : Sink: CsvTableSink(count, word)
							ship_strategy : FORWARD

{% endhighlight %}

{% top %}


text:
class PageImage < ApplicationRecord
  mount_uploader :image, ImageUploader

  belongs_to :page

  validates :page, :name, :image, presence: true
end


text:
class Seed;

class Status
{
public:
	virtual void VisitSeed_A(Seed* elm){}
	virtual void VisitSeed_B(Seed* elm){}
	virtual ~Status(){}
protected:
	Status(){}
};

class Rain_Status : public Status
{
public:
	Rain_Status(){}
	virtual ~Rain_Status(){}
	virtual void VisitSeed_A(Seed* elm);
	virtual void VisitSeed_B(Seed* elm);
};

class Sun_Status : public Status
{
public:
	Sun_Status(){}
	virtual ~Sun_Status(){}
	virtual void VisitSeed_A(Seed* elm);
	virtual void VisitSeed_B(Seed* elm);
};

#endif//STATUS_H