text:
package cluster

import (
	"bytes"
	"encoding/json"
	"flag"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"net/url"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"github.com/docker/engine-api/client"
	"github.com/docker/engine-api/types"
	"github.com/docker/engine-api/types/container"
	"github.com/docker/engine-api/types/events"
	"github.com/docker/go-connections/nat"
	"github.com/pkg/errors"
	"golang.org/x/net/context"

	"github.com/cockroachdb/cockroach/base"
	"github.com/cockroachdb/cockroach/config"
	roachClient "github.com/cockroachdb/cockroach/internal/client"
	"github.com/cockroachdb/cockroach/rpc"
	"github.com/cockroachdb/cockroach/security"
	"github.com/cockroachdb/cockroach/server"
	"github.com/cockroachdb/cockroach/util"
	"github.com/cockroachdb/cockroach/util/log"
	"github.com/cockroachdb/cockroach/util/stop"
	"github.com/cockroachdb/cockroach/util/syncutil"
	"github.com/cockroachdb/cockroach/util/timeutil"
)

const (
	builderImage     = "cockroachdb/builder"
	builderTag       = "20160804-184345"
	builderImageFull = builderImage + ":" + builderTag
	networkName      = "cockroachdb_acceptance"
)

// DefaultTCP is the default SQL/RPC port specification.
const DefaultTCP nat.Port = base.DefaultPort + "/tcp"
const defaultHTTP nat.Port = base.DefaultHTTPPort + "/tcp"

var cockroachImage = flag.String("i", builderImageFull, "the docker image to run")
var cockroachBinary = flag.String("b", defaultBinary(), "the binary to run (if image == "+builderImage+")")
var cockroachEntry = flag.String("e", "", "the entry point for the image")
var waitOnStop = flag.Bool("w", false, "wait for the user to interrupt before tearing down the cluster")
var pwd = filepath.Clean(os.ExpandEnv("${PWD}"))
var maxRangeBytes = config.DefaultZoneConfig().RangeMaxBytes

// keyLen is the length (in bits) of the generated CA and node certs.
const keyLen = 1024

func defaultBinary() string {
	gopath := filepath.SplitList(os.Getenv("GOPATH"))
	if len(gopath) == 0 {
		return ""
	}
	return gopath[0] + "/bin/docker_amd64/cockroach"
}

func exists(path string) bool {
	if _, err := os.Stat(path); os.IsNotExist(err) {
		return false
	}
	return true
}

func nodeStr(i int) string {
	return fmt.Sprintf("roach%d", i)
}

func dataStr(node, store int) string {
	return fmt.Sprintf("/data%d.%d", node, store)
}

// The various event types.
const (
	eventDie     = "die"
	eventRestart = "restart"
)

// Event for a node containing a node index and the type of event.
type Event struct {
	NodeIndex int
	Status    string
}

type testStore struct {
	index   int
	dataStr string
	config  StoreConfig
}

type testNode struct {
	*Container
	index   int
	nodeStr string
	config  NodeConfig
	stores  []testStore
}

// LocalCluster manages a local cockroach cluster running on docker. The
// cluster is composed of a "volumes" container which manages the
// persistent volumes used for certs and node data and N cockroach nodes.
type LocalCluster struct {
	client               client.APIClient
	stopper              chan struct{}
	mu                   syncutil.Mutex // Protects the fields below
	vols                 *Container
	config               TestConfig
	Nodes                []*testNode
	events               chan Event
	expectedEvents       chan Event
	oneshot              *Container
	CertsDir             string
	monitorCtx           context.Context
	monitorCtxCancelFunc func()
	logDir               string
	networkID            string
	privileged           bool // whether to run containers in privileged mode
}

// CreateLocal creates a new local cockroach cluster. The stopper is used to
// gracefully shutdown the channel (e.g. when a signal arrives). The cluster
// must be started before being used.
func CreateLocal(cfg TestConfig, logDir string, privileged bool, stopper chan struct{}) *LocalCluster {
	select {
	case <-stopper:
		// The stopper was already closed, exit early.
		os.Exit(1)
	default:
	}

	if *cockroachImage == builderImageFull && !exists(*cockroachBinary) {
		log.Fatalf(context.TODO(), "\"%s\": does not exist", *cockroachBinary)
	}

	cli, err := client.NewEnvClient()
	maybePanic(err)

	retryingClient := retryingDockerClient{
		resilientDockerClient: resilientDockerClient{APIClient: cli},
		attempts:              10,
		timeout:               10 * time.Second,
	}

	return &LocalCluster{
		client:  retryingClient,
		stopper: stopper,
		config:  cfg,
		// TODO(tschottdorf): deadlocks will occur if these channels fill up.
		events:         make(chan Event, 1000),
		expectedEvents: make(chan Event, 1000),
		logDir:         logDir,
		privileged:     privileged,
	}
}

func (l *LocalCluster) expectEvent(c *Container, msgs ...string) {
	for index, ctr := range l.Nodes {
		if c.id != ctr.id {
			continue
		}
		for _, status := range msgs {
			select {
			case l.expectedEvents <- Event{NodeIndex: index, Status: status}:
			default:
				panic("expectedEvents channel filled up")
			}
		}
		break
	}
}

// OneShot runs a container, expecting it to successfully run to completion
// and die, after which it is removed. Not goroutine safe: only one OneShot
// can be running at once.
// Adds the same binds as the cluster containers (certs, binary, etc).
func (l *LocalCluster) OneShot(
	ref string,
	ipo types.ImagePullOptions,
	containerConfig container.Config,
	hostConfig container.HostConfig,
	name string,
) error {
	if err := pullImage(l, ref, ipo); err != nil {
		return err
	}
	hostConfig.VolumesFrom = []string{l.vols.id}
	container, err := createContainer(l, containerConfig, hostConfig, name)
	if err != nil {
		return err
	}
	l.oneshot = container
	defer func() {
		if err := l.oneshot.Remove(); err != nil {
			log.Errorf(context.TODO(), "ContainerRemove: %s", err)
		}
		l.oneshot = nil
	}()

	if err := l.oneshot.Start(); err != nil {
		return err
	}
	if err := l.oneshot.Wait(); err != nil {
		return err
	}
	return nil
}

// stopOnPanic is invoked as a deferred function in Start in order to attempt
// to tear down the cluster if a panic occurs while starting it. If the panic
// was initiated by the stopper being closed (which panicOnStop notices) then
// the process is exited with a failure code.
func (l *LocalCluster) stopOnPanic() {
	if r := recover(); r != nil {
		l.stop()
		if r != l {
			panic(r)
		}
		os.Exit(1)
	}
}

// panicOnStop tests whether the stopper channel has been closed and panics if
// it has. This allows polling for whether to stop and avoids nasty locking
// complications with trying to call Stop at arbitrary points such as in the
// middle of creating a container.
func (l *LocalCluster) panicOnStop() {
	if l.stopper == nil {
		panic(l)
	}

	select {
	case <-l.stopper:
		l.stopper = nil
		panic(l)
	default:
	}
}

func (l *LocalCluster) createNetwork() {
	l.panicOnStop()

	net, err := l.client.NetworkInspect(context.Background(), networkName)
	if err == nil {
		// We need to destroy the network and any running containers inside of it.
		for containerID := range net.Containers {
			// This call could fail if the container terminated on its own after we call
			// NetworkInspect, but the likelihood of this seems low. If this line creates
			// a lot of panics we should do more careful error checking.
			maybePanic(l.client.ContainerKill(context.Background(), containerID, "9"))
		}
		maybePanic(l.client.NetworkRemove(context.Background(), networkName))
	} else if !client.IsErrNotFound(err) {
		panic(err)
	}

	resp, err := l.client.NetworkCreate(context.Background(), networkName, types.NetworkCreate{
		Driver: "bridge",
		// Docker gets very confused if two networks have the same name.
		CheckDuplicate: true,
	})
	maybePanic(err)
	if resp.Warning != "" {
		log.Warningf(context.TODO(), "creating network: %s", resp.Warning)
	}
	l.networkID = resp.ID
}

// create the volumes container that keeps all of the volumes used by
// the cluster.
func (l *LocalCluster) initCluster() {
	configJSON, err := json.Marshal(l.config)
	maybePanic(err)
	log.Infof(context.TODO(), "Initializing Cluster %s:\n%s", l.config.Name, configJSON)
	l.panicOnStop()

	// Create the temporary certs directory in the current working
	// directory. Boot2docker's handling of binding local directories
	// into the container is very confusing. If the directory being
	// bound has a parent directory that exists in the boot2docker VM
	// then that directory is bound into the container. In particular,
	// that means that binds of /tmp and /var will be problematic.
	l.CertsDir, err = ioutil.TempDir(pwd, ".localcluster.certs.")
	maybePanic(err)

	binds := []string{
		l.CertsDir + ":/certs",
		filepath.Join(pwd, "..") + ":/go/src/github.com/cockroachdb/cockroach",
	}

	if l.logDir != "" {
		if !filepath.IsAbs(l.logDir) {
			l.logDir = filepath.Join(pwd, l.logDir)
		}
		binds = append(binds, l.logDir+":/logs")
		// If we don't make sure the directory exists, Docker will and then we
		// may run into ownership issues (think Docker running as root, but us
		// running as a regular Joe as it happens on CircleCI).
		maybePanic(os.MkdirAll(l.logDir, 0777))
	}
	if *cockroachImage == builderImageFull {
		path, err := filepath.Abs(*cockroachBinary)
		maybePanic(err)
		binds = append(binds, path+":/"+filepath.Base(*cockroachBinary))
	}

	l.Nodes = []*testNode{}
	vols := map[string]struct{}{}
	// Expand the cluster configuration into nodes and stores per node.
	var nodeCount int
	for _, nc := range l.config.Nodes {
		for i := 0; i < int(nc.Count); i++ {
			newTestNode := &testNode{
				config:  nc,
				index:   nodeCount,
				nodeStr: nodeStr(nodeCount),
			}
			nodeCount++
			var storeCount int
			for _, sc := range nc.Stores {
				for j := 0; j < int(sc.Count); j++ {
					vols[dataStr(nodeCount, storeCount)] = struct{}{}
					newTestNode.stores = append(newTestNode.stores,
						testStore{
							config:  sc,
							index:   j,
							dataStr: dataStr(nodeCount, storeCount),
						})
					storeCount++
				}
			}
			l.Nodes = append(l.Nodes, newTestNode)
		}
	}

	if *cockroachImage == builderImageFull {
		maybePanic(pullImage(l, builderImage+":"+builderTag, types.ImagePullOptions{}))
	}
	c, err := createContainer(
		l,
		container.Config{
			Image:      *cockroachImage,
			Volumes:    vols,
			Entrypoint: []string{"/bin/true"},
		}, container.HostConfig{
			Binds:           binds,
			PublishAllPorts: true,
		},
		"volumes",
	)
	maybePanic(err)
	// Make sure this assignment to l.vols is before the calls to Start and Wait.
	// Otherwise, if they trigger maybePanic, this container won't get cleaned up
	// and it'll get in the way of future runs.
	l.vols = c
	maybePanic(c.Start())
	maybePanic(c.Wait())
}

func (l *LocalCluster) createRoach(node *testNode, vols *Container, env []string, cmd ...string) {
	l.panicOnStop()

	hostConfig := container.HostConfig{
		PublishAllPorts: true,
		NetworkMode:     container.NetworkMode(l.networkID),
		Privileged:      l.privileged,
	}

	if vols != nil {
		hostConfig.VolumesFrom = append(hostConfig.VolumesFrom, vols.id)
	}

	var hostname string
	if node.index >= 0 {
		hostname = fmt.Sprintf("roach%d", node.index)
	}
	var entrypoint []string
	if *cockroachImage == builderImageFull {
		entrypoint = append(entrypoint, "/"+filepath.Base(*cockroachBinary))
	} else if *cockroachEntry != "" {
		entrypoint = append(entrypoint, *cockroachEntry)
	}
	var err error
	node.Container, err = createContainer(
		l,
		container.Config{
			Hostname: hostname,
			Image:    *cockroachImage,
			ExposedPorts: map[nat.Port]struct{}{
				DefaultTCP:  {},
				defaultHTTP: {},
			},
			Entrypoint: entrypoint,
			Env:        env,
			Cmd:        cmd,
			Labels: map[string]string{
				// Allow for `docker ps --filter label=Hostname=roach0` or `--filter label=Roach`.
				"Hostname": hostname,
				"Roach":    "",
			},
		},
		hostConfig,
		node.nodeStr,
	)
	maybePanic(err)
}

func (l *LocalCluster) createCACert() {
	maybePanic(security.RunCreateCACert(
		filepath.Join(l.CertsDir, security.EmbeddedCACert),
		filepath.Join(l.CertsDir, security.EmbeddedCAKey),
		keyLen))
}

func (l *LocalCluster) createNodeCerts() {
	nodes := []string{dockerIP().String()}
	for _, node := range l.Nodes {
		nodes = append(nodes, node.nodeStr)
	}
	maybePanic(security.RunCreateNodeCert(
		filepath.Join(l.CertsDir, security.EmbeddedCACert),
		filepath.Join(l.CertsDir, security.EmbeddedCAKey),
		filepath.Join(l.CertsDir, security.EmbeddedNodeCert),
		filepath.Join(l.CertsDir, security.EmbeddedNodeKey),
		keyLen, nodes))
}

func (l *LocalCluster) startNode(node *testNode) {
	cmd := []string{
		"start",
		"--ca-cert=/certs/ca.crt",
		"--cert=/certs/node.crt",
		"--key=/certs/node.key",
		"--host=" + node.nodeStr,
		"--alsologtostderr=INFO",
		"--verbosity=1",
	}

	for _, store := range node.stores {
		storeSpec := server.StoreSpec{
			Path:        store.dataStr,
			SizeInBytes: int64(store.config.MaxRanges) * maxRangeBytes,
		}
		cmd = append(cmd, fmt.Sprintf("--store=%s", storeSpec))
	}
	// Append --join flag for all nodes except first.
	if node.index > 0 {
		cmd = append(cmd, "--join="+net.JoinHostPort(l.Nodes[0].nodeStr, base.DefaultPort))
	}

	var locallogDir string
	if len(l.logDir) > 0 {
		dockerlogDir := "/logs/" + node.nodeStr
		locallogDir = filepath.Join(l.logDir, node.nodeStr)
		maybePanic(os.MkdirAll(locallogDir, 0777))
		cmd = append(
			cmd,
			"--log-dir="+dockerlogDir,
			"--logtostderr=false")

	}
	env := []string{
		"COCKROACH_SCAN_MAX_IDLE_TIME=200ms",
		"COCKROACH_CONSISTENCY_CHECK_PANIC_ON_FAILURE=true",
		"COCKROACH_SKIP_UPDATE_CHECK=1",
	}
	l.createRoach(node, l.vols, env, cmd...)
	maybePanic(node.Start())
	httpAddr := node.Addr(defaultHTTP)

	log.Infof(context.TODO(), `*** started %[1]s ***
  ui:        %[2]s
  trace:     %[2]s/debug/requests
  logs:      %[3]s/cockroach.INFO
  pprof:     docker exec -it %[4]s /bin/bash -c 'go tool pprof /cockroach <(wget --no-check-certificate -qO- https://$(hostname):%[5]s/debug/pprof/heap)'
  cockroach: %[6]s

  cli-env:   COCKROACH_INSECURE=false COCKROACH_CA_CERT=%[7]s/ca.crt COCKROACH_CERT=%[7]s/node.crt COCKROACH_KEY=%[7]s/node.key COCKROACH_HOST=%s COCKROACH_PORT=%d`,
		node.Name(), "https://"+httpAddr.String(), locallogDir, node.Container.id[:5],
		base.DefaultHTTPPort, cmd, l.CertsDir, httpAddr.IP, httpAddr.Port)
}

func (l *LocalCluster) processEvent(event events.Message) bool {
	l.mu.Lock()
	defer l.mu.Unlock()

	// If there's currently a oneshot container, ignore any die messages from
	// it because those are expected.
	if l.oneshot != nil && event.ID == l.oneshot.id && event.Status == eventDie {
		return true
	}

	for i, n := range l.Nodes {
		if n != nil && n.id == event.ID {
			if log.V(1) {
				log.Errorf(context.TODO(), "node=%d status=%s", i, event.Status)
			}
			select {
			case l.events <- Event{NodeIndex: i, Status: event.Status}:
			default:
				panic("events channel filled up")
			}
			return true
		}
	}

	// An event on any other container is unexpected. Die.
	select {
	case <-l.stopper:
	case <-l.monitorCtx.Done():
	default:
		// There is a very tiny race here: the signal handler might be closing the
		// stopper simultaneously.
		log.Errorf(context.TODO(), "stopping due to unexpected event: %+v", event)
		if rc, err := l.client.ContainerLogs(context.Background(), event.Actor.ID, types.ContainerLogsOptions{
			ShowStdout: true,
			ShowStderr: true,
		}); err == nil {
			defer rc.Close()
			if _, err := io.Copy(os.Stderr, rc); err != nil {
				log.Infof(context.TODO(), "error listing logs: %s", err)
			}
		}
		close(l.stopper)
	}
	return false
}

func (l *LocalCluster) monitor() {
	if log.V(1) {
		log.Infof(context.TODO(), "events monitor starts")
		defer log.Infof(context.TODO(), "events monitor exits")
	}
	longPoll := func() bool {
		// If our context was cancelled, it's time to go home.
		if l.monitorCtx.Err() != nil {
			return false
		}
		rc, err := l.client.Events(l.monitorCtx, types.EventsOptions{})
		maybePanic(err)
		defer rc.Close()
		dec := json.NewDecoder(rc)
		for {
			var event events.Message
			if err := dec.Decode(&event); err != nil {
				log.Infof(context.TODO(), "event stream done, resetting...: %s", err)
				// Sometimes we get a random string-wrapped EOF error back.
				// Hard to assert on, so we just let this goroutine spin.
				return true
			}

			// Currently, the only events generated (and asserted against) are "die"
			// and "restart", to maximize compatibility across different versions of
			// Docker.
			switch event.Status {
			case eventDie, eventRestart:
			default:
				continue
			}

			if !l.processEvent(event) {
				return false
			}
		}
	}

	for longPoll() {
	}
}

// Start starts the cluster.
func (l *LocalCluster) Start() {
	defer l.stopOnPanic()

	l.mu.Lock()
	defer l.mu.Unlock()

	l.createNetwork()
	l.initCluster()
	log.Infof(context.TODO(), "creating certs (%dbit) in: %s", keyLen, l.CertsDir)
	l.createCACert()
	l.createNodeCerts()
	maybePanic(security.RunCreateClientCert(
		filepath.Join(l.CertsDir, security.EmbeddedCACert),
		filepath.Join(l.CertsDir, security.EmbeddedCAKey),
		filepath.Join(l.CertsDir, security.EmbeddedRootCert),
		filepath.Join(l.CertsDir, security.EmbeddedRootKey),
		512, security.RootUser))

	l.monitorCtx, l.monitorCtxCancelFunc = context.WithCancel(context.Background())
	go l.monitor()
	for _, node := range l.Nodes {
		l.startNode(node)
	}
}

// Assert drains the Events channel and compares the actual events with those
// expected to have been generated by the operations performed on the nodes in
// the cluster (restart, kill, ...). In the event of a mismatch, the passed
// Tester receives a fatal error.
func (l *LocalCluster) Assert(t *testing.T) {
	const almostZero = 50 * time.Millisecond
	filter := func(ch chan Event, wait time.Duration) *Event {
		select {
		case act := <-ch:
			return &act
		case <-time.After(wait):
		}
		return nil
	}

	var events []Event
	for {
		exp := filter(l.expectedEvents, almostZero)
		if exp == nil {
			break
		}
		act := filter(l.events, 15*time.Second)
		if act == nil || *exp != *act {
			t.Fatalf("expected event %v, got %v (after %v)", exp, act, events)
		}
		events = append(events, *exp)
	}
	if cur := filter(l.events, almostZero); cur != nil {
		t.Fatalf("unexpected extra event %v (after %v)", cur, events)
	}
	if log.V(2) {
		log.Infof(context.TODO(), "asserted %v", events)
	}
}

// AssertAndStop calls Assert and then stops the cluster. It is safe to stop
// the cluster multiple times.
func (l *LocalCluster) AssertAndStop(t *testing.T) {
	defer l.stop()
	l.Assert(t)
}

// stop stops the cluster.
func (l *LocalCluster) stop() {
	if *waitOnStop {
		log.Infof(context.TODO(), "waiting for interrupt")
		<-l.stopper
	}

	log.Infof(context.TODO(), "stopping")

	l.mu.Lock()
	defer l.mu.Unlock()

	if l.monitorCtxCancelFunc != nil {
		l.monitorCtxCancelFunc()
		l.monitorCtxCancelFunc = nil
	}

	if l.vols != nil {
		maybePanic(l.vols.Kill())
		maybePanic(l.vols.Remove())
		l.vols = nil
	}
	if l.CertsDir != "" {
		_ = os.RemoveAll(l.CertsDir)
		l.CertsDir = ""
	}
	outputLogDir := l.logDir
	for i, n := range l.Nodes {
		if n.Container == nil {
			continue
		}
		ci, err := n.Inspect()
		crashed := err != nil || (!ci.State.Running && ci.State.ExitCode != 0)
		maybePanic(n.Kill())
		if crashed && outputLogDir == "" {
			outputLogDir = util.CreateTempDir(util.PanicTester, "crashed_nodes")
		}
		if crashed || l.logDir != "" {
			// TODO(bdarnell): make these filenames more consistent with
			// structured logs?
			file := filepath.Join(outputLogDir, nodeStr(i),
				fmt.Sprintf("stderr.%s.log", strings.Replace(
					timeutil.Now().Format(time.RFC3339), ":", "_", -1)))

			maybePanic(os.MkdirAll(filepath.Dir(file), 0777))
			w, err := os.Create(file)
			maybePanic(err)
			defer w.Close()
			maybePanic(n.Logs(w))
			if crashed {
				log.Infof(context.TODO(), "node %d: stderr at %s", i, file)
			}
		}
		maybePanic(n.Remove())
	}
	l.Nodes = nil

	if l.networkID != "" {
		maybePanic(
			l.client.NetworkRemove(context.Background(), l.networkID))
		l.networkID = ""
	}
}

// NewClient implements the Cluster interface.
func (l *LocalCluster) NewClient(t *testing.T, i int) (*roachClient.DB, *stop.Stopper) {
	stopper := stop.NewStopper()
	rpcContext := rpc.NewContext(&base.Context{
		User:       security.NodeUser,
		SSLCA:      filepath.Join(l.CertsDir, security.EmbeddedCACert),
		SSLCert:    filepath.Join(l.CertsDir, security.EmbeddedNodeCert),
		SSLCertKey: filepath.Join(l.CertsDir, security.EmbeddedNodeKey),
	}, nil, stopper)
	sender, err := roachClient.NewSender(rpcContext, l.Nodes[i].Addr(DefaultTCP).String())
	if err != nil {
		t.Fatal(err)
	}
	return roachClient.NewDB(sender), stopper
}

// InternalIP returns the IP address used for inter-node communication.
func (l *LocalCluster) InternalIP(i int) net.IP {
	c := l.Nodes[i]
	containerInfo, err := c.Inspect()
	if err != nil {
		return nil
	}
	return net.ParseIP(containerInfo.NetworkSettings.Networks[networkName].IPAddress)
}

// PGUrl returns a URL string for the given node postgres server.
func (l *LocalCluster) PGUrl(i int) string {
	certUser := security.RootUser
	options := url.Values{}
	options.Add("sslmode", "verify-full")
	options.Add("sslcert", filepath.Join(l.CertsDir, security.EmbeddedRootCert))
	options.Add("sslkey", filepath.Join(l.CertsDir, security.EmbeddedRootKey))
	options.Add("sslrootcert", filepath.Join(l.CertsDir, security.EmbeddedCACert))
	pgURL := url.URL{
		Scheme:   "postgres",
		User:     url.User(certUser),
		Host:     l.Nodes[i].Addr(DefaultTCP).String(),
		RawQuery: options.Encode(),
	}
	return pgURL.String()
}

// NumNodes returns the number of nodes in the cluster.
func (l *LocalCluster) NumNodes() int {
	return len(l.Nodes)
}

// Kill kills the i-th node.
func (l *LocalCluster) Kill(i int) error {
	return l.Nodes[i].Kill()
}

// Restart restarts the given node. If the node isn't running, this starts it.
func (l *LocalCluster) Restart(i int) error {
	// The default timeout is 10 seconds.
	return l.Nodes[i].Restart(nil)
}

// URL returns the base url.
func (l *LocalCluster) URL(i int) string {
	return "https://" + l.Nodes[i].Addr(defaultHTTP).String()
}

// Addr returns the host and port from the node in the format HOST:PORT.
func (l *LocalCluster) Addr(i int, port string) string {
	return l.Nodes[i].Addr(nat.Port(port + "/tcp")).String()
}

// ExecRoot runs a command as root.
func (l *LocalCluster) ExecRoot(i int, cmd []string) error {
	execRoot := func(ctx context.Context) error {
		cfg := types.ExecConfig{
			User:         "root",
			Privileged:   true,
			Cmd:          cmd,
			AttachStderr: true,
			AttachStdout: true,
		}
		createResp, err := l.client.ContainerExecCreate(ctx, l.Nodes[i].Container.id, cfg)
		if err != nil {
			return err
		}
		var outputStream, errorStream bytes.Buffer
		{
			resp, err := l.client.ContainerExecAttach(ctx, createResp.ID, cfg)
			if err != nil {
				return err
			}
			defer resp.Close()
			ch := make(chan error)
			go func() {
				_, err := StdCopy(&outputStream, &errorStream, resp.Reader)
				ch <- err
			}()
			if err := <-ch; err != nil {
				return err
			}
		}
		{
			resp, err := l.client.ContainerExecInspect(ctx, createResp.ID)
			if err != nil {
				return err
			}
			if resp.Running {
				return errors.Errorf("command still running")
			}
			if resp.ExitCode != 0 {
				return fmt.Errorf("error executing %s:\n%s\n%s",
					cmd, outputStream.String(),
					errorStream.String())
			}
		}
		return nil
	}

	return retry(context.Background(), 3, 10*time.Second, "ExecRoot",
		matchNone, execRoot)
}


text:
var BaseReporter = require('./Base');
var util = require('util');

var escTCString = function (message) {
  if(!message) {
    return '';
  }

  return message.
      replace(/\|/g, '||').
    	replace(/\'/g, '|\'').
    	replace(/\n/g, '|n').
    	replace(/\r/g, '|r').
    	replace(/\u0085/g, '|x').
    	replace(/\u2028/g, '|l').
    	replace(/\u2029/g, '|p').
    	replace(/\[/g, '|[').
    	replace(/\]/g, '|]');
};

var getTestName = function (result) {
  return result.slice(1).join(' ');
}

var TeamcityReporter = function(formatError, reportSlow) {
  BaseReporter.call(this, formatError, reportSlow);

  this.onRunStart = function(browsers) {
    this._browsers = browsers;
    this.browserResults = {};
  };

  this.writeCommonMsg = function(msg) {
    this.write('\n' + msg);
    this._dotsCount = 0;
  };

  this.specSuccess = function(browser, result) {
    var browseResult = this.checkNewSuit(browser, result);
    var testName = getTestName(result.suite);

    browseResult.log.push(util.format(this.TEST_START, escTCString(testName)));
    browseResult.log.push(util.format(this.TEST_END,
      escTCString(testName), result.time));
  };

  this.specFailure = function(browser, result) {
    var browseResult = this.checkNewSuit(browser, result);
    var testName = getTestName(result.suite);

    browseResult.log.push(util.format(this.TEST_START, escTCString(testName)));
    browseResult.log.push(util.format(this.TEST_FAILED, escTCString(testName),
      escTCString(JSON.stringify(result.log))));
    browseResult.log.push(util.format(this.TEST_END,
      escTCString(testName), result.time));
  };

  this.specSkipped = function(browser, result) {
    var browseResult = this.checkNewSuit(browser, result);
    var testName = getTestName(result.suite);

    browseResult.log.push(util.format(this.TEST_IGNORED, escTCString(testName)));
  };

  this.checkNewSuit = function(browser, result) {
    var browserResult = this.checkNewBrowser(browser);
    var suiteExists = browserResult.suits.indexOf(result.suite[0]) !== -1;

    if(!suiteExists) {
      if(browserResult.suits.length > 0) {
        browserResult.log.push(util.format(this.SUITE_END,
          escTCString(browserResult.suits[browserResult.suits.length - 1])));
      }
      browserResult.suits.push(result.suite[0]);
      browserResult.log.push(util.format(this.SUITE_START, escTCString(result.suite[0])));
    }
    return browserResult;
  };

  this.checkNewBrowser = function(browser) {
    if(!this.browserResults[browser.id]) {
      this.browserResults[browser.id] = {
        log : [],
        suits : []
      };
    }
    return this.browserResults[browser.id];
  };

  this.onRunComplete = function(browsers, results) {
    var self = this;

    Object.keys(this.browserResults).forEach(function(key) {
      var browserResult = self.browserResults[key];
      if(browserResult.suits.length > 0) {
        browserResult.log.push(util.format(self.SUITE_END,
          escTCString(browserResult.suits[browserResult.suits.length - 1])));
      }
      self.write(self.BROWSER_START, key);
      self.write(browserResult.log.join('\n'));
      self.write(self.BROWSER_END, key);
    });

    this.writeCommonMsg(browsers.map(this.renderBrowser).join('\n') + '\n');

    if (browsers.length > 1 && !results.disconnected && !results.error) {
      if (!results.failed) {
        this.write(this.TOTAL_SUCCESS, results.success);
      } else {
        this.write(this.TOTAL_FAILED, results.failed, results.success);
      }
    }
  };

  this.TEST_IGNORED = '##teamcity[testIgnored name=\'%s\']';
  this.SUITE_START = '##teamcity[testSuiteStarted name=\'%s\']';
  this.SUITE_END = '##teamcity[testSuiteFinished name=\'%s\']';
  this.TEST_START = '##teamcity[testStarted name=\'%s\']';
  this.TEST_FAILED = '##teamcity[testFailed name=\'%s\' message=\'FAILED\' details=\'%s\']';
  this.TEST_END = '##teamcity[testFinished name=\'%s\' duration=\'%s\']';
  this.BROWSER_START = '##teamcity[browserStart name=\'%s\']\n';
  this.BROWSER_END = '\n##teamcity[browserEnd name=\'%s\']';
};

module.exports = TeamcityReporter;

redpajama_set_name: RedPajamaGithub


text:
package org.lumi.chaptertwo.jintro;

import java.util.Scanner;

/**
 * Created by John Tsantilis (A.K.A lumi) on 30/1/2017.
 */

public class Arithmetic {
    public static void main(String[] args) {
        int num1;
        int num2;
        Scanner input = new Scanner(System.in);

        System.out.println("Enter first integer: ");

        num1 = input.nextInt();

        System.out.println("Enter second integer: ");
        num2 = input.nextInt();

        System.out.printf("%s%d%n", "Sum is: ", num1 + num2);
        System.out.printf("%s%d%n", "Product is: ", num1 * num2);
        System.out.printf("%s%d%n", "Quotient is: ", num1 / num2);

    }

}


text:
package json

import (
	"encoding/json"
	"io"

	"gopkg.in/Clever/optimus.v3"
)

// New writes all of the Rows in a Table as newline-separate JSON objects.
func New(out io.Writer) optimus.Sink {
	return func(source optimus.Table) error {
		defer source.Stop()
		for row := range source.Rows() {
			obj, err := json.Marshal(row)
			if err != nil {
				return err
			}
			obj = append(obj, byte('\n'))
			if _, err := out.Write(obj); err != nil {
				return err
			}
		}
		if source.Err() != nil {
			return source.Err()
		}
		return nil
	}
}


text:
package org.apache.geronimo.web.info;

import java.io.Serializable;

/**
 * @version $Rev:$ $Date:$
 */
public class ErrorPageInfo implements Serializable {

    public String location;
    public Integer errorCode;
    public String exceptionType;

}


text:
@class NSCollectionView, NSCollectionViewLayout, NSIndexPath;

@protocol NSCollectionViewDelegateFlowLayout <NSCollectionViewDelegate>

@optional
- (struct CGSize)collectionView:(NSCollectionView *)arg1 layout:(NSCollectionViewLayout *)arg2 referenceSizeForFooterInSection:(long long)arg3;
- (struct CGSize)collectionView:(NSCollectionView *)arg1 layout:(NSCollectionViewLayout *)arg2 referenceSizeForHeaderInSection:(long long)arg3;
- (double)collectionView:(NSCollectionView *)arg1 layout:(NSCollectionViewLayout *)arg2 minimumInteritemSpacingForSectionAtIndex:(long long)arg3;
- (double)collectionView:(NSCollectionView *)arg1 layout:(NSCollectionViewLayout *)arg2 minimumLineSpacingForSectionAtIndex:(long long)arg3;
- (struct NSEdgeInsets)collectionView:(NSCollectionView *)arg1 layout:(NSCollectionViewLayout *)arg2 insetForSectionAtIndex:(long long)arg3;
- (struct CGSize)collectionView:(NSCollectionView *)arg1 layout:(NSCollectionViewLayout *)arg2 sizeForItemAtIndexPath:(NSIndexPath *)arg3;
@end


text:
#ifndef SkImage_Base_DEFINED
#define SkImage_Base_DEFINED

#include "SkImage.h"

class SkImage_Base : public SkImage {
public:
    SkImage_Base(int width, int height) : INHERITED(width, height) {}

    virtual void onDraw(SkCanvas*, SkScalar x, SkScalar y, const SkPaint*) = 0;
    virtual void onDrawRectToRect(SkCanvas*, const SkRect* src,
                                  const SkRect& dst, const SkPaint*) = 0;

    // Default impl calls onDraw
    virtual bool onReadPixels(SkBitmap*, const SkIRect& subset) const;

    virtual const void* onPeekPixels(SkImageInfo*, size_t* /*rowBytes*/) const {
        return NULL;
    }

    virtual GrTexture* onGetTexture() { return NULL; }

    // return a read-only copy of the pixels. We promise to not modify them,
    // but only inspect them (or encode them).
    virtual bool getROPixels(SkBitmap*) const { return false; }

    virtual SkShader* onNewShader(SkShader::TileMode, SkShader::TileMode) const { return NULL; };
private:
    typedef SkImage INHERITED;
};

#endif


text:
using UnityEngine;
using System.Collections.Generic;
using System;

namespace Assets.Scripts.Rendering
{
	[RequireComponent(typeof(MeshFilter))]
	[RequireComponent(typeof(PolygonCollider2D))]
	[ExecuteInEditMode]
	public class Mesh2DColliderMaker : MonoBehaviour
	{
		MeshFilter filter;
		PolygonCollider2D polyCollider;

		#region Unity Callers

		void Start()
		{
			filter = GetComponent<MeshFilter>();
			polyCollider = GetComponent<PolygonCollider2D>();
		}

#if UNITY_EDITOR
		void Update()
		{
			if (!Application.isPlaying)
				CreatePolygon2DColliderPoints();
		}
#endif

		#endregion

		#region API

		public void CreatePolygon2DColliderPoints()
		{
			var edges = BuildEdgesFromMesh();
			var paths = BuildColliderPaths(edges);
			ApplyPathsToPolygonCollider(paths);
		}

		#endregion

		#region Helper

		void ApplyPathsToPolygonCollider(List<Vector2[]> paths)
		{
			if (paths == null)
				return;

			polyCollider.pathCount = paths.Count;
			for (int i = 0; i < paths.Count; i++)
			{
				var path = paths[i];
				polyCollider.SetPath(i, path);
			}
		}

		Dictionary<Edge2D, int> BuildEdgesFromMesh()
		{
			var mesh = filter.sharedMesh;

			if (mesh == null)
				return null;

			var verts = mesh.vertices;
			var tris = mesh.triangles;
			var edges = new Dictionary<Edge2D, int>();

			for (int i = 0; i < tris.Length - 2; i += 3)
			{

				var faceVert1 = verts[tris[i]];
				var faceVert2 = verts[tris[i + 1]];
				var faceVert3 = verts[tris[i + 2]];

				if (Math.Round(faceVert1.z, 3) != 0 || Math.Round(faceVert2.z, 3) != 0 || Math.Round(faceVert3.z, 3) != 0)
					continue;

				Edge2D[] faceEdges;
				faceEdges = new Edge2D[] {
				new Edge2D{ a = faceVert1, b = faceVert2 },
				new Edge2D{ a = faceVert2, b = faceVert3 },
				new Edge2D{ a = faceVert3, b = faceVert1 },
			};

				foreach (var edge in faceEdges)
				{
					if (edges.ContainsKey(edge))
						edges[edge]++;
					else
						edges[edge] = 1;
				}
			}

			return edges;
		}

		static List<Edge2D> GetOuterEdges(Dictionary<Edge2D, int> allEdges)
		{
			var outerEdges = new List<Edge2D>();

			foreach (var edge in allEdges.Keys)
			{
				var numSharedFaces = allEdges[edge];
				if (numSharedFaces == 1)
					outerEdges.Add(edge);
			}

			return outerEdges;
		}

		static List<Vector2[]> BuildColliderPaths(Dictionary<Edge2D, int> allEdges)
		{

			if (allEdges == null)
				return null;

			var outerEdges = GetOuterEdges(allEdges);

			var paths = new List<List<Edge2D>>();
			List<Edge2D> path = null;

			while (outerEdges.Count > 0)
			{

				if (path == null)
				{
					path = new List<Edge2D>();
					path.Add(outerEdges[0]);
					paths.Add(path);

					outerEdges.RemoveAt(0);
				}

				bool foundAtLeastOneEdge = false;

				int i = 0;
				while (i < outerEdges.Count)
				{
					var edge = outerEdges[i];
					bool removeEdgeFromOuter = false;

					if (edge.b == path[0].a)
					{
						path.Insert(0, edge);
						removeEdgeFromOuter = true;
					}
					else if (edge.a == path[path.Count - 1].b)
					{
						path.Add(edge);
						removeEdgeFromOuter = true;
					}

					if (removeEdgeFromOuter)
					{
						foundAtLeastOneEdge = true;
						outerEdges.RemoveAt(i);
					}
					else
						i++;
				}

				//If we didn't find at least one edge, then the remaining outer edges must belong to a different path
				if (!foundAtLeastOneEdge)
					path = null;

			}

			var cleanedPaths = new List<Vector2[]>();

			foreach (var builtPath in paths)
			{
				var coords = new List<Vector2>();

				foreach (var edge in builtPath)
					coords.Add(edge.a);

				cleanedPaths.Add(CoordinatesCleaned(coords));
			}

			return cleanedPaths;
		}

		static bool CoordinatesFormLine(Vector2 a, Vector2 b, Vector2 c)
		{
			//If the area of a triangle created from three points is zero, they must be in a line.
			float area = a.x * (b.y - c.y) +
				b.x * (c.y - a.y) +
					c.x * (a.y - b.y);

			return Mathf.Approximately(area, 0f);

		}

		static Vector2[] CoordinatesCleaned(List<Vector2> coordinates)
		{
			List<Vector2> coordinatesCleaned = new List<Vector2>();
			coordinatesCleaned.Add(coordinates[0]);

			var lastAddedIndex = 0;

			for (int i = 1; i < coordinates.Count; i++)
			{

				var coordinate = coordinates[i];

				Vector2 lastAddedCoordinate = coordinates[lastAddedIndex];
				Vector2 nextCoordinate = (i + 1 >= coordinates.Count) ? coordinates[0] : coordinates[i + 1];

				if (!CoordinatesFormLine(lastAddedCoordinate, coordinate, nextCoordinate))
				{

					coordinatesCleaned.Add(coordinate);
					lastAddedIndex = i;

				}

			}

			return coordinatesCleaned.ToArray();

		}

		#endregion

		#region Nested

		struct Edge2D
		{

			public Vector2 a;
			public Vector2 b;

			public override bool Equals(object obj)
			{
				if (obj is Edge2D)
				{
					var edge = (Edge2D)obj;
					//An edge is equal regardless of which order it's points are in
					return (edge.a == a && edge.b == b) || (edge.b == a && edge.a == b);
				}

				return false;

			}

			public override int GetHashCode()
			{
				return a.GetHashCode() ^ b.GetHashCode();
			}

			public override string ToString()
			{
				return string.Format("[" + a.x + "," + a.y + "->" + b.x + "," + b.y + "]");
			}

		}
		#endregion
	}
}


text:
package org.assertj.core.util.introspection;

import static java.lang.reflect.Modifier.isStatic;
import static org.assertj.core.util.Preconditions.checkArgument;

import java.lang.reflect.Field;
import java.lang.reflect.Modifier;

/**
 * Shameless copy from Apache commons lang and then modified to keep only the interesting stuff for AssertJ.
 *
 * Utilities for working with fields by reflection. Adapted and refactored from the dormant [reflect] Commons sandbox
 * component.
 *

 * The ability is provided to break the scoping restrictions coded by the programmer. This can allow fields to be
 * changed that shouldn't be. This facility should be used with care.
 */
class FieldUtils {

  /**
   * Gets an accessible Field by name breaking scope if requested. Superclasses/interfaces will be
   * considered.
   *
   * @param cls the class to reflect, must not be null
   * @param fieldName the field name to obtain
   * @param forceAccess whether to break scope restrictions using the setAccessible method.
   *          False will only match public fields.
   * @return the Field object
   * @throws IllegalArgumentException if the class or field name is null
   * @throws IllegalAccessException if field exists but is not public
   */
  static Field getField(final Class<?> cls, String fieldName, boolean forceAccess) throws IllegalAccessException {
    checkArgument(cls != null, "The class must not be null");
    checkArgument(fieldName != null, "The field name must not be null");
    // Sun Java 1.3 has a bugged implementation of getField hence we write the
    // code ourselves

    // getField() will return the Field object with the declaring class
    // set correctly to the class that declares the field. Thus requesting the
    // field on a subclass will return the field from the superclass.
    //
    // priority order for lookup:
    // searchclass private/protected/package/public
    // superclass protected/package/public
    // private/different package blocks access to further superclasses
    // implementedinterface public

    // check up the superclass hierarchy
    for (Class<?> acls = cls; acls != null; acls = acls.getSuperclass()) {
      try {
        Field field = acls.getDeclaredField(fieldName);
        // getDeclaredField checks for non-public scopes as well and it returns accurate results
        if (!Modifier.isPublic(field.getModifiers())) {
          if (forceAccess) {
            field.setAccessible(true);
          } else {
            throw new IllegalAccessException("can not access" + fieldName + " because it is not public");
          }
        }
        return field;
      } catch (NoSuchFieldException ex) { // NOPMD
        // ignore
      }
    }
    // check the public interface case. This must be manually searched for
    // incase there is a public supersuperclass field hidden by a private/package
    // superclass field.
    Field match = null;
    for (Class<?> class1 : ClassUtils.getAllInterfaces(cls)) {
      try {
        Field test = class1.getField(fieldName);
        checkArgument(match == null, "Reference to field " + fieldName + " is ambiguous relative to " + cls
                                     + "; a matching field exists on two or more implemented interfaces.");
        match = test;
      } catch (NoSuchFieldException ex) { // NOPMD
        // ignore
      }
    }
    return match;
  }

  /**
   * Reads an accessible Field.
   *
   * @param field the field to use
   * @param target the object to call on, may be null for static fields
   * @return the field value
   * @throws IllegalArgumentException if the field is null
   * @throws IllegalAccessException if the field is not accessible
   */
  private static Object readField(Field field, Object target) throws IllegalAccessException {
    return readField(field, target, false);
  }

  /**
   * Reads a Field.
   *
   * @param field the field to use
   * @param target the object to call on, may be null for static fields
   * @param forceAccess whether to break scope restrictions using the setAccessible method.
   * @return the field value
   * @throws IllegalArgumentException if the field is null
   * @throws IllegalAccessException if the field is not made accessible
   */
  private static Object readField(Field field, Object target, boolean forceAccess) throws IllegalAccessException {
    checkArgument(field != null, "The field must not be null");
    if (forceAccess && !field.isAccessible()) {
      field.setAccessible(true);
    } else {
      MemberUtils.setAccessibleWorkaround(field);
    }
    return field.get(target);
  }

  /**
   * Reads the named field. Superclasses will be considered.
   *

   * Since 3.19.0 static and synthetic fields are ignored.
   *
   * @param target the object to reflect, must not be null
   * @param fieldName the field name to obtain
   * @param forceAccess whether to break scope restrictions using the setAccessible method.
   *          False will only match public fields.
   * @return the field value
   * @throws IllegalArgumentException if the class or field name is null or the field can not be found.
   * @throws IllegalAccessException if the named field is not made accessible
   */
  static Object readField(Object target, String fieldName, boolean forceAccess) throws IllegalAccessException {
    checkArgument(target != null, "target object must not be null");
    Class<?> cls = target.getClass();
    Field field = getField(cls, fieldName, forceAccess);
    checkArgument(field != null, "Cannot locate field %s on %s", fieldName, cls);
    checkArgument(!isStatic(field.getModifiers()), "Reading static field is not supported and field %s is static on %s",
                  fieldName, cls);
    checkArgument(!field.isSynthetic(), "Reading synthetic field is not supported and field %s is", fieldName);
    // already forced access above, don't repeat it here:
    return readField(field, target);
  }

}


text:
package org.apache.spark.sql.execution.columnar

import org.scalatest.BeforeAndAfterEach

import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSQLContext
import org.apache.spark.sql.test.SQLTestData._

class PartitionBatchPruningSuite
  extends SparkFunSuite
  with BeforeAndAfterEach
  with SharedSQLContext {

  import testImplicits._

  private lazy val originalColumnBatchSize = spark.conf.get(SQLConf.COLUMN_BATCH_SIZE)
  private lazy val originalInMemoryPartitionPruning =
    spark.conf.get(SQLConf.IN_MEMORY_PARTITION_PRUNING)
  private val testArrayData = (1 to 100).map { key =>
    Tuple1(Array.fill(key)(key))
  }
  private val testBinaryData = (1 to 100).map { key =>
    Tuple1(Array.fill(key)(key.toByte))
  }

  override protected def beforeAll(): Unit = {
    super.beforeAll()
    // Make a table with 5 partitions, 2 batches per partition, 10 elements per batch
    spark.conf.set(SQLConf.COLUMN_BATCH_SIZE.key, 10)
    // Enable in-memory partition pruning
    spark.conf.set(SQLConf.IN_MEMORY_PARTITION_PRUNING.key, true)
    // Enable in-memory table scan accumulators
    spark.conf.set(SQLConf.IN_MEMORY_TABLE_SCAN_STATISTICS_ENABLED.key, "true")
  }

  override protected def afterAll(): Unit = {
    try {
      spark.conf.set(SQLConf.COLUMN_BATCH_SIZE.key, originalColumnBatchSize)
      spark.conf.set(SQLConf.IN_MEMORY_PARTITION_PRUNING.key, originalInMemoryPartitionPruning)
    } finally {
      super.afterAll()
    }
  }

  override protected def beforeEach(): Unit = {
    super.beforeEach()
    // This creates accumulators, which get cleaned up after every single test,
    // so we need to do this before every test.
    val pruningData = sparkContext.makeRDD((1 to 100).map { key =>
      val string = if (((key - 1) / 10) % 2 == 0) null else key.toString
      TestData(key, string)
    }, 5).toDF()
    pruningData.createOrReplaceTempView("pruningData")
    spark.catalog.cacheTable("pruningData")

    val pruningStringData = sparkContext.makeRDD((100 to 200).map { key =>
      StringData(key.toString)
    }, 5).toDF()
    pruningStringData.createOrReplaceTempView("pruningStringData")
    spark.catalog.cacheTable("pruningStringData")

    val pruningArrayData = sparkContext.makeRDD(testArrayData, 5).toDF()
    pruningArrayData.createOrReplaceTempView("pruningArrayData")
    spark.catalog.cacheTable("pruningArrayData")

    val pruningBinaryData = sparkContext.makeRDD(testBinaryData, 5).toDF()
    pruningBinaryData.createOrReplaceTempView("pruningBinaryData")
    spark.catalog.cacheTable("pruningBinaryData")
  }

  override protected def afterEach(): Unit = {
    try {
      spark.catalog.uncacheTable("pruningData")
      spark.catalog.uncacheTable("pruningStringData")
      spark.catalog.uncacheTable("pruningArrayData")
      spark.catalog.uncacheTable("pruningBinaryData")
    } finally {
      super.afterEach()
    }
  }

  // Comparisons
  checkBatchPruning("SELECT key FROM pruningData WHERE key = 1", 1, 1)(Seq(1))
  checkBatchPruning("SELECT key FROM pruningData WHERE 1 = key", 1, 1)(Seq(1))
  checkBatchPruning("SELECT key FROM pruningData WHERE key <=> 1", 1, 1)(Seq(1))
  checkBatchPruning("SELECT key FROM pruningData WHERE 1 <=> key", 1, 1)(Seq(1))
  checkBatchPruning("SELECT key FROM pruningData WHERE key < 12", 1, 2)(1 to 11)
  checkBatchPruning("SELECT key FROM pruningData WHERE key <= 11", 1, 2)(1 to 11)
  checkBatchPruning("SELECT key FROM pruningData WHERE key > 88", 1, 2)(89 to 100)
  checkBatchPruning("SELECT key FROM pruningData WHERE key >= 89", 1, 2)(89 to 100)
  checkBatchPruning("SELECT key FROM pruningData WHERE 12 > key", 1, 2)(1 to 11)
  checkBatchPruning("SELECT key FROM pruningData WHERE 11 >= key", 1, 2)(1 to 11)
  checkBatchPruning("SELECT key FROM pruningData WHERE 88 < key", 1, 2)(89 to 100)
  checkBatchPruning("SELECT key FROM pruningData WHERE 89 <= key", 1, 2)(89 to 100)
  // Do not filter on array type
  checkBatchPruning("SELECT _1 FROM pruningArrayData WHERE _1 = array(1)", 5, 10)(Seq(Array(1)))
  checkBatchPruning("SELECT _1 FROM pruningArrayData WHERE _1 <= array(1)", 5, 10)(Seq(Array(1)))
  checkBatchPruning("SELECT _1 FROM pruningArrayData WHERE _1 >= array(1)", 5, 10)(
    testArrayData.map(_._1))
  // Do not filter on binary type
  checkBatchPruning(
    "SELECT _1 FROM pruningBinaryData WHERE _1 == binary(chr(1))", 5, 10)(Seq(Array(1.toByte)))

  // IS NULL
  checkBatchPruning("SELECT key FROM pruningData WHERE value IS NULL", 5, 5) {
    (1 to 10) ++ (21 to 30) ++ (41 to 50) ++ (61 to 70) ++ (81 to 90)
  }

  // IS NOT NULL
  checkBatchPruning("SELECT key FROM pruningData WHERE value IS NOT NULL", 5, 5) {
    (11 to 20) ++ (31 to 40) ++ (51 to 60) ++ (71 to 80) ++ (91 to 100)
  }

  // Conjunction and disjunction
  checkBatchPruning("SELECT key FROM pruningData WHERE key > 8 AND key <= 21", 2, 3)(9 to 21)
  checkBatchPruning("SELECT key FROM pruningData WHERE key < 2 OR key > 99", 2, 2)(Seq(1, 100))
  checkBatchPruning("SELECT key FROM pruningData WHERE key < 12 AND key IS NOT NULL", 1, 2)(1 to 11)
  checkBatchPruning("SELECT key FROM pruningData WHERE key < 2 OR (key > 78 AND key < 92)", 3, 4) {
    Seq(1) ++ (79 to 91)
  }
  checkBatchPruning("SELECT key FROM pruningData WHERE NOT (key < 88)", 1, 2) {
    // Although the `NOT` operator isn't supported directly, the optimizer can transform
    // `NOT (a < b)` to `b >= a`
    88 to 100
  }

  // Support `IN` predicate
  checkBatchPruning("SELECT key FROM pruningData WHERE key IN (1)", 1, 1)(Seq(1))
  checkBatchPruning("SELECT key FROM pruningData WHERE key IN (1, 2)", 1, 1)(Seq(1, 2))
  checkBatchPruning("SELECT key FROM pruningData WHERE key IN (1, 11)", 1, 2)(Seq(1, 11))
  checkBatchPruning("SELECT key FROM pruningData WHERE key IN (1, 21, 41, 61, 81)", 5, 5)(
    Seq(1, 21, 41, 61, 81))
  checkBatchPruning("SELECT CAST(s AS INT) FROM pruningStringData WHERE s = '100'", 1, 1)(Seq(100))
  checkBatchPruning("SELECT CAST(s AS INT) FROM pruningStringData WHERE s < '102'", 1, 1)(
    Seq(100, 101))
  checkBatchPruning(
    "SELECT CAST(s AS INT) FROM pruningStringData WHERE s IN ('99', '150', '201')", 1, 1)(
      Seq(150))
  // Do not filter on array type
  checkBatchPruning("SELECT _1 FROM pruningArrayData WHERE _1 IN (array(1), array(2, 2))", 5, 10)(
    Seq(Array(1), Array(2, 2)))

  // With unsupported `InSet` predicate
  {
    val seq = (1 to 30).mkString(", ")
    checkBatchPruning(s"SELECT key FROM pruningData WHERE key IN ($seq)", 5, 10)(1 to 30)
    checkBatchPruning(s"SELECT key FROM pruningData WHERE NOT (key IN ($seq))", 5, 10)(31 to 100)
    checkBatchPruning(s"SELECT key FROM pruningData WHERE NOT (key IN ($seq)) AND key > 88", 1, 2) {
      89 to 100
    }
  }

  // Support `StartsWith` predicate
  checkBatchPruning("SELECT CAST(s AS INT) FROM pruningStringData WHERE s like '18%'", 1, 1)(
    180 to 189
  )
  checkBatchPruning("SELECT CAST(s AS INT) FROM pruningStringData WHERE s like '%'", 5, 11)(
    100 to 200
  )
  checkBatchPruning("SELECT CAST(s AS INT) FROM pruningStringData WHERE '18%' like s", 5, 11)(Seq())

  // With disable IN_MEMORY_PARTITION_PRUNING option
  test("disable IN_MEMORY_PARTITION_PRUNING") {
    spark.conf.set(SQLConf.IN_MEMORY_PARTITION_PRUNING.key, false)

    val df = sql("SELECT key FROM pruningData WHERE key = 1")
    val result = df.collect().map(_(0)).toArray
    assert(result.length === 1)

    val (readPartitions, readBatches) = df.queryExecution.executedPlan.collect {
        case in: InMemoryTableScanExec => (in.readPartitions.value, in.readBatches.value)
      }.head
    assert(readPartitions === 5)
    assert(readBatches === 10)
  }

  def checkBatchPruning(
      query: String,
      expectedReadPartitions: Int,
      expectedReadBatches: Int)(
      expectedQueryResult: => Seq[Any]): Unit = {

    test(query) {
      val df = sql(query)
      val queryExecution = df.queryExecution

      assertResult(expectedQueryResult.toArray, s"Wrong query result: $queryExecution") {
        df.collect().map(_(0)).toArray
      }

      val (readPartitions, readBatches) = df.queryExecution.executedPlan.collect {
        case in: InMemoryTableScanExec => (in.readPartitions.value, in.readBatches.value)
      }.head

      assert(readBatches === expectedReadBatches, s"Wrong number of read batches: $queryExecution")
      assert(
        readPartitions === expectedReadPartitions,
        s"Wrong number of read partitions: $queryExecution")
    }
  }
}


text:
<?xml version="1.0" encoding="UTF-8"?>
<project version="4">
  <component name="GradleSettings">
    <option name="linkedExternalProjectsSettings">
      <GradleProjectSettings>
        <option name="distributionType" value="DEFAULT_WRAPPED" />
        <option name="externalProjectPath" value="$PROJECT_DIR$" />
        <option name="gradleJvm" value="1.8" />
        <option name="modules">
          <set>
            <option value="$PROJECT_DIR$" />
            <option value="$PROJECT_DIR$/ProToi" />
            <option value="$PROJECT_DIR$/app" />
            <option value="$PROJECT_DIR$/circleimageview" />
            <option value="$PROJECT_DIR$/core" />
            <option value="$PROJECT_DIR$/libraryCircularPager" />
            <option value="$PROJECT_DIR$/library___" />
            <option value="$PROJECT_DIR$/library_image_loader" />
            <option value="$PROJECT_DIR$/library_scrolling" />
            <option value="$PROJECT_DIR$/librarycColorArt" />
            <option value="$PROJECT_DIR$/observablescrollview" />
            <option value="$PROJECT_DIR$/sample___" />
            <option value="$PROJECT_DIR$/simpleInfiniteCarouselmaster" />
            <option value="$PROJECT_DIR$/youTubeAndroidAPIDemo" />
          </set>
        </option>
      </GradleProjectSettings>
    </option>
  </component>
</project>


text:
<!DOCTYPE HTML>

    <meta charset="utf-8">
    <title>DateTimePicker - sap.ui.webc.main</title>


text:
package org.thunlp.mapred;

import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapred.*;

public class MapReduceJobConf extends JobConf {
    public MapReduceJobConf() {
        super();
        this.setInputFormat(SequenceFileInputFormat.class);
        this.setOutputFormat(SequenceFileOutputFormat.class);
        this.setNumMapTasks(48);
        this.setNumReduceTasks(12);
    }

    public MapReduceJobConf(Class jobClass) {
        super(jobClass);
        this.setInputFormat(SequenceFileInputFormat.class);
        this.setOutputFormat(SequenceFileOutputFormat.class);
        this.setNumMapTasks(48);
        this.setNumReduceTasks(12);
    }

    public void setKeyValueClass(
            Class<? extends WritableComparable> keyMerge,
            Class<? extends Writable> valueMerge,
            Class<? extends WritableComparable> keyOut,
            Class<? extends Writable> valueOut) {
        this.setMapOutputKeyClass(keyMerge);
        this.setMapOutputValueClass(valueMerge);
        this.setOutputKeyClass(keyOut);
        this.setOutputValueClass(valueOut);
    }

    public void setMapReduce(
            Class<? extends Mapper> mapper,
            Class<? extends Reducer> reducer) {
        this.setMapperClass(mapper);
        this.setReducerClass(reducer);
    }
}


text:
package eu.paweld2.ao.architecture.common.datatype;

import lombok.Data;

import java.io.Serializable;
import java.util.List;

@Data
public class AoTreeNode<T extends Serializable> implements Serializable {
    private final T data;
    private final List<AoTreeNode<T>> children;

    public AoTreeNode(T data, List<AoTreeNode<T>> children) {
        this.data = data;
        this.children = children;
    }

}


text:
namespace qe {
    namespace render {
        namespace vulkan {
            class DLL_EXPORT VulkanBuffer : public VulkanObject {
            public:
                vk::Device device;
                vk::Buffer buffer;
                vk::DeviceMemory memory;
                vk::DescriptorBufferInfo descriptor;

                vk::DeviceSize size = 0;
                vk::DeviceSize alignment = 0;
                void* mapped = nullptr;

                /// Usage flags to be filled by external source at buffer creation (to query at some later point) */
                vk::BufferUsageFlags usageFlags;
                /// Memory propertys flags to be filled by external source at buffer creation (to query at some later point) */
                vk::MemoryPropertyFlagBits memoryPropertyFlags;

            public:
                explicit VulkanBuffer(vk::DeviceSize device_size);
                ~VulkanBuffer();

                /// Map a memory range of this buffer. If successful, mapped points to the specified buffer range.
                ///
                /// @param size (Optional) Size of the memory range to map. Pass VK_WHOLE_SIZE to map the complete buffer range.
                /// @param offset (Optional) Byte offset from beginning
                ///
                /// @return vk::Result  of the buffer mapping call
                vk::Result  map(vk::DeviceSize size = VK_WHOLE_SIZE, vk::DeviceSize offset = 0);

                /// Unmap a mapped memory range
                ///
                /// @note Does not return a result as vkUnmapMemory can't fail
                void unmap();

                /// Attach the allocated memory block to the buffer
                ///
                /// @param offset (Optional) Byte offset (from the beginning) for the memory region to bind
                ///
                /// @return vk::Result  of the bindBufferMemory call
                vk::Result  bind(vk::DeviceSize offset = 0);

                /// Setup the default descriptor for this buffer
                ///
                /// @param size (Optional) Size of the memory range of the descriptor
                /// @param offset (Optional) Byte offset from beginning
                ///
                void setupDescriptor(vk::DeviceSize size = VK_WHOLE_SIZE, vk::DeviceSize offset = 0);

                /// Copies the specified data to the mapped buffer
                ///
                /// @param data Pointer to the data to copy
                /// @param size Size of the data to copy in machine units
                void copyTo(void* data, vk::DeviceSize size);

                /// Flush a memory range of the buffer to make it visible to the device
                ///
                /// @note Only required for non-coherent memory
                ///
                /// @param size (Optional) Size of the memory range to flush. Pass VK_WHOLE_SIZE to flush the complete buffer range.
                /// @param offset (Optional) Byte offset from beginning
                ///
                /// @return vk::Result  of the flush call
                vk::Result  flush(vk::DeviceSize size = VK_WHOLE_SIZE, vk::DeviceSize offset = 0);

                /// Invalidate a memory range of the buffer to make it visible to the host
                ///
                /// @note Only required for non-coherent memory
                ///
                /// @param size (Optional) Size of the memory range to invalidate. Pass VK_WHOLE_SIZE to invalidate the complete buffer range.
                /// @param offset (Optional) Byte offset from beginning
                ///
                /// @return vk::Result  of the invalidate call
                vk::Result  invalidate(vk::DeviceSize size = VK_WHOLE_SIZE, vk::DeviceSize offset = 0);

                /// Release all Vulkan resources held by this buffer
                void destroy();

            };
        }
    }
}


text:
package org.kuali.rice.ksb.messaging.serviceexporters;

import org.kuali.rice.ksb.api.bus.ServiceDefinition;

import javax.xml.namespace.QName;

/**
 * Manages service endpoints that are exported from this application
 *
 * @author Kuali Rice Team (rice.collab@kuali.org)
 */
public interface ServiceExportManager {

    Object getService(QName serviceName);

    QName getServiceName(String url);

    void exportService(ServiceDefinition serviceDefinition);

    void removeService(QName serviceName);

}


text:
<?php

declare(strict_types=1);

namespace phpDocumentor\Descriptor\Example;

use Mockery\Adapter\Phpunit\MockeryTestCase;
use org\bovigo\vfs\vfsStream;
use phpDocumentor\Descriptor\Tag\ExampleDescriptor;
use Symfony\Component\Filesystem\Filesystem;
use function chdir;
use function sys_get_temp_dir;

/**
 * Tests for the \phpDocumentor\Descriptor\Example\Finder class.
 */
final class FinderTest extends MockeryTestCase
{
    public const EXAMPLE_TEXT = 'This is an example';

    /** @var Filesystem */
    private $filesystem;

    /** @var Finder */
    private $fixture;

    /**
     * Initializes the fixture.
     */
    protected function setUp() : void
    {
        $this->filesystem = new Filesystem();
        $this->fixture = new Finder();
    }

    /**
     * @covers \phpDocumentor\Descriptor\Example\Finder::setSourceDirectory
     * @covers \phpDocumentor\Descriptor\Example\Finder::getSourceDirectory
     */
    public function testGetAndSetSourceDirectory() : void
    {
        $this->assertSame('', $this->fixture->getSourceDirectory());

        $this->fixture->setSourceDirectory('this/is/a/test');

        $this->assertSame('this/is/a/test', $this->fixture->getSourceDirectory());
    }

    /**
     * @covers \phpDocumentor\Descriptor\Example\Finder::setExampleDirectories
     * @covers \phpDocumentor\Descriptor\Example\Finder::getExampleDirectories
     */
    public function testGetAndSetExampleDirectories() : void
    {
        $this->assertSame([], $this->fixture->getExampleDirectories());

        $this->fixture->setExampleDirectories(['this/is/a/test']);

        $this->assertSame(['this/is/a/test'], $this->fixture->getExampleDirectories());
    }

    /**
     * @covers \phpDocumentor\Descriptor\Example\Finder::find
     * @covers \phpDocumentor\Descriptor\Example\Finder::getExampleFileContents
     * @covers \phpDocumentor\Descriptor\Example\Finder::constructExamplePath
     */
    public function testFindExampleContentsInExampleDirectory() : void
    {
        $directories = [vfsStream::url('root/exampleDirectory'), vfsStream::url('root/exampleDirectory2')];

        $descriptor = $this->givenADescriptorWithExamplePath('example.txt');
        $this->givenTheDirectoryStructure(
            [
                'exampleDirectory' => [],
                'exampleDirectory2' => ['example.txt' => self::EXAMPLE_TEXT],
                'source' => ['example.txt' => 'this is not it'], // check if the example directory precedes this
            ]
        );

        $this->fixture->setExampleDirectories($directories);
        $this->fixture->setSourceDirectory(vfsStream::url('root/source'));
        $result = $this->fixture->find($descriptor);

        $this->assertSame(self::EXAMPLE_TEXT, $result);
    }

    /**
     * @covers \phpDocumentor\Descriptor\Example\Finder::find
     * @covers \phpDocumentor\Descriptor\Example\Finder::getExampleFileContents
     * @covers \phpDocumentor\Descriptor\Example\Finder::getExamplePathFromSource
     */
    public function testFindExampleContentsInSourceDirectory() : void
    {
        $descriptor = $this->givenADescriptorWithExamplePath('example.txt');
        $this->givenTheDirectoryStructure(['source' => ['example.txt' => self::EXAMPLE_TEXT]]);

        $this->fixture->setSourceDirectory(vfsStream::url('root/source'));
        $result = $this->fixture->find($descriptor);

        $this->assertSame(self::EXAMPLE_TEXT, $result);
    }

    /**
     * @covers \phpDocumentor\Descriptor\Example\Finder::find
     * @covers \phpDocumentor\Descriptor\Example\Finder::getExampleFileContents
     * @covers \phpDocumentor\Descriptor\Example\Finder::getExamplePathFromExampleDirectory
     */
    public function testFindExampleContentsInExamplesDirectoryOfWorkingDirectory() : void
    {
        // can't use vfsStream because we are working from the Current Working Directory, which is not
        // supported by vfsStream
        $workingDirectory = sys_get_temp_dir() . '/phpdoc-tests';
        $this->givenExampleFileInFolder($workingDirectory . '/examples/example.txt');

        $descriptor = $this->givenADescriptorWithExamplePath('example.txt');

        chdir($workingDirectory);
        $result = $this->fixture->find($descriptor);

        $this->assertSame(self::EXAMPLE_TEXT, $result);

        chdir('..');
        $this->filesystem->remove($workingDirectory);
    }

    /**
     * @covers \phpDocumentor\Descriptor\Example\Finder::find
     * @covers \phpDocumentor\Descriptor\Example\Finder::getExampleFileContents
     */
    public function testFindExampleContentsInCurrentWorkingDirectory() : void
    {
        // can't use vfsStream because we are working from the Current Working Directory, which is not
        // supported by vfsStream
        $workingDirectory = sys_get_temp_dir() . '/phpdoc-tests';
        $this->givenExampleFileInFolder($workingDirectory . '/example.txt');

        $descriptor = $this->givenADescriptorWithExamplePath('example.txt');

        chdir($workingDirectory);
        $result = $this->fixture->find($descriptor);

        $this->assertSame(self::EXAMPLE_TEXT, $result);

        chdir('..');
        $this->filesystem->remove($workingDirectory);
    }

    /**
     * @covers \phpDocumentor\Descriptor\Example\Finder::find
     * @covers \phpDocumentor\Descriptor\Example\Finder::getExampleFileContents
     */
    public function testErrorMessageIsReturnedIfFileIsNotFound() : void
    {
        $filename = 'doesNotExist.txt';
        $descriptor = $this->givenADescriptorWithExamplePath($filename);

        $result = $this->fixture->find($descriptor);

        $this->assertSame('** File not found : ' . $filename . ' **', $result);
    }

    /**
     * Returns an ExampleDescriptor with the given filename set.
     */
    private function givenADescriptorWithExamplePath(string $path) : ExampleDescriptor
    {
        $descriptor = new ExampleDescriptor('example');
        $descriptor->setFilePath($path);

        return $descriptor;
    }

    /**
     * Initializes a virtual folder structure used to verify file io operations.
     *
     * @param string[] $structure
     */
    private function givenTheDirectoryStructure(array $structure) : void
    {
        vfsStream::setup('root', null, $structure);
    }

    /**
     * Creates an example file at the given path and creates folders where necessary.
     */
    private function givenExampleFileInFolder(string $exampleFilename) : void
    {
        $this->filesystem->dumpFile($exampleFilename, self::EXAMPLE_TEXT);
    }
}


text:
ACCEPTED

#### According to
Index Fungorum

#### Published in
null

#### Original name
Parodiella spegazzinii var. indigoferae Marchal & Steyaert

### Remarks
null


text:
using System.Collections.Generic;
using osu.Game.Beatmaps;
using osu.Game.Rulesets.Mods;
using osu.Game.Rulesets.Osu.UI;
using osu.Game.Rulesets.UI;
using osuTK;

namespace osu.Game.Rulesets.Osu.Edit
{
    public class DrawableOsuEditRuleset : DrawableOsuRuleset
    {
        public DrawableOsuEditRuleset(Ruleset ruleset, IWorkingBeatmap beatmap, IReadOnlyList<Mod> mods)
            : base(ruleset, beatmap, mods)
        {
        }

        protected override Playfield CreatePlayfield() => new OsuPlayfieldNoCursor();

        public override PlayfieldAdjustmentContainer CreatePlayfieldAdjustmentContainer() => new OsuPlayfieldAdjustmentContainer { Size = Vector2.One };

        private class OsuPlayfieldNoCursor : OsuPlayfield
        {
            protected override GameplayCursorContainer CreateCursor() => null;
        }
    }
}


text:
package org.apache.flink.runtime.source.coordinator;

import org.apache.flink.api.connector.source.Boundedness;
import org.apache.flink.api.connector.source.mocks.MockSource;
import org.apache.flink.api.connector.source.mocks.MockSourceSplit;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.coordination.MockOperatorCoordinatorContext;
import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
import org.apache.flink.runtime.operators.coordination.RecreateOnResetOperatorCoordinator;
import org.apache.flink.runtime.source.event.ReaderRegistrationEvent;

import org.junit.Before;
import org.junit.Test;

import java.util.concurrent.CompletableFuture;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;

/**
 * Unit tests for {@link SourceCoordinatorProvider}.
 */
public class SourceCoordinatorProviderTest {
	private static final OperatorID OPERATOR_ID = new OperatorID(1234L, 5678L);
	private static final int NUM_SPLITS = 10;
	private SourceCoordinatorProvider<MockSourceSplit> provider;

	@Before
	public void setup() {
		provider = new SourceCoordinatorProvider<>(
				"SourceCoordinatorProviderTest",
				OPERATOR_ID,
				new MockSource(Boundedness.BOUNDED, NUM_SPLITS),
				1);
	}

	@Test
	public void testCreate() {
		OperatorCoordinator coordinator =
				provider.create(new MockOperatorCoordinatorContext(OPERATOR_ID, NUM_SPLITS));
		assertTrue(coordinator instanceof RecreateOnResetOperatorCoordinator);
	}

	@Test
	public void testCheckpointAndReset() throws Exception {
		final OperatorCoordinator.Context context = new MockOperatorCoordinatorContext(OPERATOR_ID, NUM_SPLITS);
		final RecreateOnResetOperatorCoordinator coordinator =
				(RecreateOnResetOperatorCoordinator) provider.create(context);
		final SourceCoordinator<?, ?> sourceCoordinator =
				(SourceCoordinator<?, ?>) coordinator.getInternalCoordinator();

		// Start the coordinator.
		coordinator.start();
		// register reader 0 and take a checkpoint.
		coordinator.handleEventFromOperator(0, new ReaderRegistrationEvent(0, "location"));
		CompletableFuture<byte[]> future = new CompletableFuture<>();
		coordinator.checkpointCoordinator(0L, future);
		byte[] bytes = future.get();

		// Register reader 1.
		coordinator.handleEventFromOperator(1, new ReaderRegistrationEvent(1, "location"));
		// Wait until the coordinator context is updated with registration of reader 1.
		while (sourceCoordinator.getContext().registeredReaders().size() < 2) {
			Thread.sleep(1);
		}

		// reset the coordinator to the checkpoint which only contains reader 0.
		coordinator.resetToCheckpoint(bytes);
		final SourceCoordinator<?, ?> restoredSourceCoordinator =
				(SourceCoordinator<?, ?>) coordinator.getInternalCoordinator();
		assertNotEquals("The restored source coordinator should be a different instance",
				restoredSourceCoordinator, sourceCoordinator);
		assertEquals("There should only be one registered reader.",
				1, restoredSourceCoordinator.getContext().registeredReaders().size());
		assertNotNull("The only registered reader should be reader 0",
				restoredSourceCoordinator.getContext().registeredReaders().get(0));
	}

}


text:
package com.baidu.palo.master;

import com.baidu.palo.catalog.Catalog;
import com.baidu.palo.catalog.Database;
import com.baidu.palo.catalog.MaterializedIndex;
import com.baidu.palo.catalog.OlapTable;
import com.baidu.palo.catalog.Partition;
import com.baidu.palo.catalog.Replica;
import com.baidu.palo.catalog.Replica.ReplicaState;
import com.baidu.palo.catalog.Tablet;
import com.baidu.palo.catalog.TabletInvertedIndex;
import com.baidu.palo.clone.CloneChecker;
import com.baidu.palo.common.MetaNotFoundException;
import com.baidu.palo.common.util.Daemon;
import com.baidu.palo.persist.ReplicaPersistInfo;
import com.baidu.palo.system.Backend;
import com.baidu.palo.task.AgentBatchTask;
import com.baidu.palo.task.AgentTask;
import com.baidu.palo.task.AgentTaskExecutor;
import com.baidu.palo.task.AgentTaskQueue;
import com.baidu.palo.task.DropReplicaTask;
import com.baidu.palo.task.MasterTask;
import com.baidu.palo.task.PushTask;
import com.baidu.palo.task.StorageMediaMigrationTask;
import com.baidu.palo.thrift.TBackend;
import com.baidu.palo.thrift.TDisk;
import com.baidu.palo.thrift.TMasterResult;
import com.baidu.palo.thrift.TPushType;
import com.baidu.palo.thrift.TReportRequest;
import com.baidu.palo.thrift.TStatus;
import com.baidu.palo.thrift.TStatusCode;
import com.baidu.palo.thrift.TStorageMedium;
import com.baidu.palo.thrift.TTablet;
import com.baidu.palo.thrift.TTabletInfo;
import com.baidu.palo.thrift.TTaskType;

import com.google.common.collect.LinkedListMultimap;
import com.google.common.collect.ListMultimap;
import com.google.common.collect.Lists;
import com.google.common.collect.Queues;

import org.apache.commons.lang.StringUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.thrift.TException;

import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;

public class ReportHandler extends Daemon {
    private static final Logger LOG = LogManager.getLogger(ReportHandler.class);

    private BlockingQueue<ReportTask> reportQueue = Queues.newLinkedBlockingQueue();

    public ReportHandler() {
    }

    public TMasterResult handleReport(TReportRequest request) throws TException {
        TMasterResult result = new TMasterResult();
        TStatus tStatus = new TStatus(TStatusCode.OK);
        result.setStatus(tStatus);

        // get backend
        TBackend tBackend = request.getBackend();
        String host = tBackend.getHost();
        int bePort = tBackend.getBe_port();
        Backend backend = Catalog.getCurrentSystemInfo().getBackendWithBePort(host, bePort);
        if (backend == null) {
            tStatus.setStatus_code(TStatusCode.INTERNAL_ERROR);
            List<String> errorMsgs = Lists.newArrayList();
            errorMsgs.add("backend[" + host + ":" + bePort + "] does not exist.");
            tStatus.setError_msgs(errorMsgs);
            return result;
        }

        long beId = backend.getId();
        Map<TTaskType, Set<Long>> tasks = null;
        Map<String, TDisk> disks = null;
        Map<Long, TTablet> tablets = null;
        long reportVersion = -1;

        if (request.isSetTasks()) {
            tasks = request.getTasks();
        }
        if (request.isSetDisks()) {
            disks = request.getDisks();
        }
        if (request.isSetTablets()) {
            tablets = request.getTablets();
            reportVersion = request.getReport_version();
        }

        ReportTask reportTask = new ReportTask(beId, tasks, disks, tablets, reportVersion);

        try {
            reportQueue.put(reportTask);
        } catch (InterruptedException e) {
            tStatus.setStatus_code(TStatusCode.INTERNAL_ERROR);
            List<String> errorMsgs = Lists.newArrayList();
            errorMsgs.add("failed to put report task to queue. queue size: " + reportQueue.size());
            errorMsgs.add("err: " + e.getMessage());
            tStatus.setError_msgs(errorMsgs);
            return result;
        }
        LOG.info("receive report from be {}. current queue size: {}", backend.getId(), reportQueue.size());
        return result;
    }

    private class ReportTask extends MasterTask {
        private long beId;
        private Map<TTaskType, Set<Long>> tasks;
        private Map<String, TDisk> disks;
        private Map<Long, TTablet> tablets;
        private long reportVersion;

        public ReportTask(long beId, Map<TTaskType, Set<Long>> tasks,
                Map<String, TDisk> disks,
                Map<Long, TTablet> tablets, long reportVersion) {
            this.beId = beId;
            this.tasks = tasks;
            this.disks = disks;
            this.tablets = tablets;
            this.reportVersion = reportVersion;
        }

        @Override
        protected void exec() {
            if (tasks != null) {
                ReportHandler.taskReport(beId, tasks);
            }

            if (disks != null) {
                ReportHandler.diskReport(beId, disks);
            }

            if (tablets != null) {
                long backendReportVersion = Catalog.getCurrentSystemInfo().getBackendReportVersion(beId);
                if (reportVersion < backendReportVersion) {
                    LOG.warn("out of date report version {} from backend[{}]. current report version[{}]",
                             reportVersion, beId, backendReportVersion);
                } else {
                    ReportHandler.tabletReport(beId, tablets, reportVersion);
                }
            }
        }
    }

    private static void taskReport(long backendId, Map<TTaskType, Set<Long>> runningTasks) {
        LOG.info("begin to handle task report from backend {}", backendId);
        long start = System.currentTimeMillis();

        for (TTaskType type : runningTasks.keySet()) {
            Set<Long> taskSet = runningTasks.get(type);
            if (!taskSet.isEmpty()) {
                String signatures = StringUtils.join(taskSet, ", ");
                LOG.debug("backend task[{}]: {}", type.name(), signatures);
            }
        }

        List<AgentTask> diffTasks = AgentTaskQueue.getDiffTasks(backendId, runningTasks);

        AgentBatchTask batchTask = new AgentBatchTask();
        for (AgentTask task : diffTasks) {
            // these tasks donot need to do diff
            // 1. CREATE
            // 2. SYNC DELETE
            // 3. CHECK_CONSISTENCY
            if (task.getTaskType() == TTaskType.CREATE
                    || (task.getTaskType() == TTaskType.PUSH && ((PushTask) task).getPushType() == TPushType.DELETE
                    && ((PushTask) task).isSyncDelete())
                    || task.getTaskType() == TTaskType.CHECK_CONSISTENCY) {
                continue;
            }
            batchTask.addTask(task);
        }

        LOG.debug("get {} diff task(s) to resend", batchTask.getTaskNum());
        if (batchTask.getTaskNum() > 0) {
            AgentTaskExecutor.submit(batchTask);
        }

        LOG.info("finished to handle task report from backend {}, diff task num: {}. cost: {} ms",
                 backendId, batchTask.getTaskNum(), (System.currentTimeMillis() - start));
    }

    private static void diskReport(long backendId, Map<String, TDisk> backendDisks) {
        LOG.info("begin to handle disk report from backend {}", backendId);
        long start = System.currentTimeMillis();

        Backend backend = Catalog.getCurrentSystemInfo().getBackend(backendId);
        if (backend == null) {
            LOG.warn("backend doesn't exist. id: " + backendId);
            return;
        }

        backend.updateDisks(backendDisks);
        LOG.info("finished to handle disk report from backend {}, cost: {} ms",
                 backendId, (System.currentTimeMillis() - start));
    }

    private static void tabletReport(long backendId, Map<Long, TTablet> backendTablets, long backendReportVersion) {
        LOG.info("begin to handle tablet report from backend {}, tablet num: {}, report version: {}",
                 backendId, backendTablets.size(), backendReportVersion);
        long start = System.currentTimeMillis();

        // storage medium map
        HashMap<Long, TStorageMedium> storageMediumMap = Catalog.getInstance().getPartitionIdToStorageMediumMap();

        // db id -> tablet id
        ListMultimap<Long, Long> tabletSyncMap = LinkedListMultimap.create();
        // db id -> tablet id
        ListMultimap<Long, Long> tabletDeleteFromMeta = LinkedListMultimap.create();
        // tablet ids which schema hash is valid
        Set<Long> foundTabletsWithValidSchema = new HashSet<Long>();
        // tablet ids which schema hash is invalid
        Map<Long, TTabletInfo> foundTabletsWithInvalidSchema = new HashMap<Long, TTabletInfo>();
        // storage medium -> tablet id
        ListMultimap<TStorageMedium, Long> tabletMigrationMap = LinkedListMultimap.create();

        // 1. do the diff. find out (intersection) / (be - meta) / (meta - be)
        Catalog.getCurrentInvertedIndex().tabletReport(backendId, backendTablets, storageMediumMap,
                                                       tabletSyncMap,
                                                       tabletDeleteFromMeta,
                                                       foundTabletsWithValidSchema,
                                                       foundTabletsWithInvalidSchema,
                                                       tabletMigrationMap);

        // 2. sync
        sync(backendTablets, tabletSyncMap, backendId, backendReportVersion);

        // 3. delete (meta - be)
        // BE will automatically drop defective tablets. these tablets should also be dropped in catalog
        deleteFromMeta(tabletDeleteFromMeta, backendId, backendReportVersion);

        // 4. handle (be - meta)
        deleteFromBackend(backendTablets, foundTabletsWithValidSchema, foundTabletsWithInvalidSchema, backendId);

        // 5. migration (ssd <-> hdd)
        handleMigration(tabletMigrationMap, backendId);

        LOG.info("finished to handle tablet report from backend {}, cost: {} ms",
                 backendId, (System.currentTimeMillis() - start));
    }

    private static void sync(Map<Long, TTablet> backendTablets, ListMultimap<Long, Long> tabletSyncMap,
                             long backendId, long backendReportVersion) {
        TabletInvertedIndex invertedIndex = Catalog.getCurrentInvertedIndex();
        for (Long dbId : tabletSyncMap.keySet()) {
            Database db = Catalog.getInstance().getDb(dbId);
            if (db == null) {
                continue;
            }
            db.writeLock();
            try {
                int syncCounter = 0;
                List<Long> tabletIds = tabletSyncMap.get(dbId);
                LOG.info("before sync tablets in db[{}]. report num: {}. backend[{}]",
                         dbId, tabletIds.size(), backendId);
                for (Long tabletId : tabletIds) {
                    long tableId = invertedIndex.getTableId(tabletId);
                    OlapTable olapTable = (OlapTable) db.getTable(tableId);
                    if (olapTable == null) {
                        continue;
                    }

                    long partitionId = invertedIndex.getPartitionId(tabletId);
                    Partition partition = olapTable.getPartition(partitionId);
                    if (partition == null) {
                        continue;
                    }

                    long indexId = invertedIndex.getIndexId(tabletId);
                    MaterializedIndex index = partition.getIndex(indexId);
                    if (index == null) {
                        continue;
                    }
                    long schemaHash = olapTable.getSchemaHashByIndexId(indexId);

                    Tablet tablet = index.getTablet(tabletId);
                    if (tablet == null) {
                        continue;
                    }

                    Replica replica = tablet.getReplicaByBackendId(backendId);
                    if (replica == null) {
                        continue;
                    }

                    if (replica.getState() == ReplicaState.NORMAL) {
                        long metaVersion = replica.getVersion();
                        long metaVersionHash = replica.getVersionHash();
                        long backendVersion = -1L;
                        long backendVersionHash = -1L;
                        long rowCount = -1L;
                        long dataSize = -1L;

                        for (TTabletInfo tabletInfo : backendTablets.get(tabletId).getTablet_infos()) {
                            if (tabletInfo.getSchema_hash() == schemaHash) {
                                backendVersion = tabletInfo.getVersion();
                                backendVersionHash = tabletInfo.getVersion_hash();
                                rowCount = tabletInfo.getRow_count();
                                dataSize = tabletInfo.getData_size();
                                break;
                            }
                        }
                        if (backendVersion == -1L || backendVersionHash == -1L) {
                            continue;
                        }

                        if (metaVersion < backendVersion
                                || (metaVersion == backendVersion && metaVersionHash != backendVersionHash)) {

                            if (backendReportVersion < Catalog.getCurrentSystemInfo()
                                    .getBackendReportVersion(backendId)) {
                                continue;
                            }

                            // happens when PUSH finished in BE but failed or not yet report to FE
                            replica.updateInfo(backendVersion, backendVersionHash, dataSize, rowCount);

                            ++syncCounter;
                            LOG.debug("sync replica[{}] in db[{}].", replica.getId(), dbId);
                        } else {
                            LOG.debug("replica[{}] version is changed between check and real sync."
                                    + " meta[{}-{}]. backend[{}-{}]",
                                      replica.getId(), metaVersion, metaVersionHash,
                                      backendVersion, backendVersionHash);
                        }
                    }
                } // end for tabletMetaSyncMap
                LOG.info("sync {} tablets in db[{}]. backend[{}]", syncCounter, dbId, backendId);
            } finally {
                db.writeUnlock();
            }
        } // end for dbs
    }

    private static void deleteFromMeta(ListMultimap<Long, Long> tabletDeleteFromMeta, long backendId,
                                       long backendReportVersion) {
        TabletInvertedIndex invertedIndex = Catalog.getCurrentInvertedIndex();
        for (Long dbId : tabletDeleteFromMeta.keySet()) {
            Database db = Catalog.getInstance().getDb(dbId);
            if (db == null) {
                continue;
            }
            db.writeLock();
            try {
                int deleteCounter = 0;
                List<Long> tabletIds = tabletDeleteFromMeta.get(dbId);
                for (Long tabletId : tabletIds) {
                    long tableId = invertedIndex.getTableId(tabletId);
                    OlapTable olapTable = (OlapTable) db.getTable(tableId);
                    if (olapTable == null) {
                        continue;
                    }

                    long partitionId = invertedIndex.getPartitionId(tabletId);
                    Partition partition = olapTable.getPartition(partitionId);
                    if (partition == null) {
                        continue;
                    }

                    short replicationNum = olapTable.getPartitionInfo().getReplicationNum(partition.getId());

                    long indexId = invertedIndex.getIndexId(tabletId);
                    MaterializedIndex index = partition.getIndex(indexId);
                    if (index == null) {
                        continue;
                    }

                    Tablet tablet = index.getTablet(tabletId);
                    if (tablet == null) {
                        continue;
                    }

                    Replica replica = tablet.getReplicaByBackendId(backendId);
                    if (replica == null) {
                        continue;
                    }

                    // check report version again
                    if (backendReportVersion < Catalog.getCurrentSystemInfo().getBackendReportVersion(backendId)) {
                        continue;
                    }

                    ReplicaState state = replica.getState();
                    if (state == ReplicaState.NORMAL || state == ReplicaState.SCHEMA_CHANGE) {
                        // if state is PENDING / ROLLUP / CLONE
                        // it's normal that the replica is not created in BE but exists in meta.
                        // so we do not delete it.
                        List<Replica> replicas = tablet.getReplicas();
                        if (replicas.size() <= 1) {
                            LOG.error("invalid situation. tablet[{}] has few replica[{}]", tabletId, replicas.size());
                            continue;
                        }

                        tablet.deleteReplicaByBackendId(backendId);
                        ++deleteCounter;

                        // handle related task
                        Catalog.getInstance().handleJobsWhenDeleteReplica(tableId, partitionId, indexId, tabletId,
                                                                          replica.getId(), backendId);

                        // write edit log
                        ReplicaPersistInfo info = ReplicaPersistInfo.createForDelete(dbId, tableId, partitionId,
                                                                                indexId, tabletId, backendId);

                        Catalog.getInstance().getEditLog().logDeleteReplica(info);
                        LOG.warn("delete replica[{}] in tablet[{}] from meta. backend[{}]",
                                 replica.getId(), tabletId, backendId);

                        // check for clone
                        replicas = tablet.getReplicas();
                        if (replicas.size() == 0) {
                            LOG.error("invalid situation. tablet[{}] is empty", tabletId);
                        } else if (replicas.size() < replicationNum) {
                            CloneChecker.getInstance().checkTabletForSupplement(dbId, tableId,
                                                                                partitionId,
                                                                                indexId, tabletId);
                        }
                    }
                } // end for tabletMetas
                LOG.info("delete {} replica(s) from catalog in db[{}]", deleteCounter, dbId);
            } finally {
                db.writeUnlock();
            }
        } // end for dbs
    }

    private static void deleteFromBackend(Map<Long, TTablet> backendTablets,
                                          Set<Long> foundTabletsWithValidSchema,
                                          Map<Long, TTabletInfo> foundTabletsWithInvalidSchema,
                                          long backendId) {
        int deleteFromBackendCounter = 0;
        int addToMetaCounter = 0;
        AgentBatchTask batchTask = new AgentBatchTask();
        for (Long tabletId : backendTablets.keySet()) {
            TTablet backendTablet = backendTablets.get(tabletId);

            for (TTabletInfo backendTabletInfo : backendTablet.getTablet_infos()) {
                boolean needDelete = false;
                if (!foundTabletsWithValidSchema.contains(tabletId)) {
                    // if this tablet is not in meta. try adding it.
                    // if add failed. delete this tablet from backend.
                    try {
                        addReplica(tabletId, backendTabletInfo, backendId);
                        // update counter
                        needDelete = false;
                        ++addToMetaCounter;
                    } catch (MetaNotFoundException e) {
                        LOG.warn("failed add to meta. tablet[{}], backend[{}]. {}",
                                 tabletId, backendId, e.getMessage());
                        needDelete = true;
                    }
                }

                if (needDelete) {
                    // drop replica
                    DropReplicaTask task = new DropReplicaTask(backendId, tabletId, backendTabletInfo.getSchema_hash());
                    batchTask.addTask(task);
                    LOG.warn("delete tablet[" + tabletId + " - " + backendTabletInfo.getSchema_hash()
                            + "] from backend[" + backendId + "] because not found in meta");
                    ++deleteFromBackendCounter;
                }
            } // end for tabletInfos

            if (foundTabletsWithInvalidSchema.containsKey(tabletId)) {
                // this tablet is found in meta but with invalid schema hash.
                // delete it.
                int schemaHash = foundTabletsWithInvalidSchema.get(tabletId).getSchema_hash();
                DropReplicaTask task = new DropReplicaTask(backendId, tabletId, schemaHash);
                batchTask.addTask(task);
                LOG.warn("delete tablet[" + tabletId + " - " + schemaHash + "] from backend[" + backendId
                        + "] because invalid schema hash");
                ++deleteFromBackendCounter;
            }
        } // end for backendTabletIds
        AgentTaskExecutor.submit(batchTask);

        LOG.info("delete {} tablet(s) from backend[{}]", deleteFromBackendCounter, backendId);
        LOG.info("add {} replica(s) to meta. backend[{}]", addToMetaCounter, backendId);
    }

    private static void handleMigration(ListMultimap<TStorageMedium, Long> tabletMetaMigrationMap,
                                        long backendId) {
        TabletInvertedIndex invertedIndex = Catalog.getCurrentInvertedIndex();
        AgentBatchTask batchTask = new AgentBatchTask();
        for (TStorageMedium storageMedium : tabletMetaMigrationMap.keySet()) {
            List<Long> tabletIds = tabletMetaMigrationMap.get(storageMedium);
            for (Long tabletId : tabletIds) {
                StorageMediaMigrationTask task = new StorageMediaMigrationTask(backendId, tabletId,
                                                      invertedIndex.getEffectiveSchemaHash(tabletId),
                                                      storageMedium);
                batchTask.addTask(task);
            }
        }

        AgentTaskExecutor.submit(batchTask);
    }

    private static void addReplica(long tabletId, TTabletInfo backendTabletInfo, long backendId)
            throws MetaNotFoundException {
        TabletInvertedIndex invertedIndex = Catalog.getCurrentInvertedIndex();
        long dbId = invertedIndex.getDbId(tabletId);
        long tableId = invertedIndex.getTableId(tabletId);
        long partitionId = invertedIndex.getPartitionId(tabletId);
        long indexId = invertedIndex.getIndexId(tabletId);

        int schemaHash = backendTabletInfo.getSchema_hash();
        long version = backendTabletInfo.getVersion();
        long versionHash = backendTabletInfo.getVersion_hash();
        long dataSize = backendTabletInfo.getData_size();
        long rowCount = backendTabletInfo.getRow_count();

        Database db = Catalog.getInstance().getDb(dbId);
        if (db == null) {
            throw new MetaNotFoundException("db[" + dbId + "] does not exist");
        }
        db.writeLock();
        try {
            OlapTable olapTable = (OlapTable) db.getTable(tableId);
            if (olapTable == null) {
                throw new MetaNotFoundException("table[" + tableId + "] does not exist");
            }

            Partition partition = olapTable.getPartition(partitionId);
            if (partition == null) {
                throw new MetaNotFoundException("partition[" + partitionId + "] does not exist");
            }
            short replicationNum = olapTable.getPartitionInfo().getReplicationNum(partition.getId());

            MaterializedIndex materializedIndex = partition.getIndex(indexId);
            if (materializedIndex == null) {
                throw new MetaNotFoundException("index[" + indexId + "] does not exist");
            }
            Tablet tablet = materializedIndex.getTablet(tabletId);
            if (tablet == null) {
                throw new MetaNotFoundException("tablet[" + tabletId + "] does not exist");
            }

            long committedVersion = partition.getCommittedVersion();
            long committedVersionHash = partition.getCommittedVersionHash();

            // check replica version
            if (version < committedVersion || (version == committedVersion && versionHash != committedVersionHash)) {
                throw new MetaNotFoundException("version is invalid. tablet[" + version + "-" + versionHash + "]"
                        + ", committed[" + committedVersion + "-" + committedVersionHash + "]");
            }

            // check schema hash
            if (schemaHash != olapTable.getSchemaHashByIndexId(indexId)) {
                throw new MetaNotFoundException("schema hash is diff[" + schemaHash + "-"
                        + olapTable.getSchemaHashByIndexId(indexId) + "]");
            }

            List<Replica> replicas = tablet.getReplicas();
            int replicationOnLine = 0;
            for (Replica replica : replicas) {
                final long id = replica.getBackendId();
                final Backend backend = Catalog.getCurrentSystemInfo().getBackend(id);
                if (backend.isAlive() && !backend.isDecommissioned() && replica.getState() == ReplicaState.NORMAL) {
                    replicationOnLine++;
                }
            }

            if (replicationOnLine < replicationNum) {
                long replicaId = Catalog.getInstance().getNextId();
                Replica replica = new Replica(replicaId, backendId, version, versionHash,
                                              dataSize, rowCount, ReplicaState.NORMAL);
                tablet.addReplica(replica);

                // write edit log
                ReplicaPersistInfo info = ReplicaPersistInfo.createForAdd(dbId, tableId, partitionId, indexId,
                                                                          tabletId, backendId, replicaId,
                                                                          version, versionHash, dataSize, rowCount);

                Catalog.getInstance().getEditLog().logAddReplica(info);

                LOG.info("add replica[{}-{}] to catalog. backend[{}]", tabletId, replicaId, backendId);
            } else {
                // replica is enough. check if this tablet is already in meta
                // (status changed between 'tabletReport()' and 'addReplica()')
                for (Replica replica : replicas) {
                    if (replica.getBackendId() == backendId) {
                        // tablet is already in meta. return true
                        return;
                    }
                }

                throw new MetaNotFoundException("replica is enough[" + replicas.size() + "-" + replicationNum + "]");
            }
        } finally {
            db.writeUnlock();
        }
    }

    @Override
    protected void runOneCycle() {
        while (true) {
            ReportTask task = null;
            try {
                task = reportQueue.take();
                task.exec();
            } catch (InterruptedException e) {
                LOG.warn("got interupted exception when executing report", e);
                continue;
            }
        }
    }
}


text:
r"""A small Django app that provides template tags for Markdown using the
python-markdown library.

See <http://github.com/douzepouze/django-markdown-tag> for more info.
"""

__version_info__ = (1, 0, 0)
__version__ = '.'.join(map(str, __version_info__))
__author__ = "Steffen Görtz"

def markdown(text, options={}):
    from markdown_tag.conf import settings
    import markdown
    if not text:
        return ""
    local_options = settings.MARKDOWN_OPTIONS_DEFAULT
    local_options.update(options)
    return markdown.markdown(text, **local_options)